broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['visible_now'][0] >= r['visible_now'][1]) and (r['not_expired'][0] < r['not_expired'][1]) and (r['consumer_credit'][0] < r['consumer_credit'][1]) and (r['group_head'][0] == min(r['group_head'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'visible_now': [10, 10], 'not_expired': [9, 10], 'no_live_reservation': [8, 8], 'consumer_credit': [2, 3], 'group_head': [3, [3, 4]]}
check('valid operation', solve(r), True)
check('A scheduled message is delivered before its visibility instant', solve(dict(r, **{'visible_now': [9, 10]})), False)
check('A message is delivered after its expiration deadline', solve(dict(r, **{'not_expired': [10, 10]})), False)
check('A second worker reserves a currently leased message', solve(dict(r, **{'no_live_reservation': [9, 8]})), False)
check('The broker exceeds consumer in-flight credit', solve(dict(r, **{'consumer_credit': [3, 3]})), False)
check('A FIFO message overtakes an earlier group member', solve(dict(r, **{'group_head': [4, [3, 4]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'visible_now': [9, 10], 'not_expired': [10, 10], 'no_live_reservation': [9, 8], 'consumer_credit': [3, 3], 'group_head': [4, [3, 4]]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-reservation

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['visible_now'][0] >= r['visible_now'][1]) and (r['not_expired'][0] < r['not_expired'][1]) and (r['no_live_reservation'][0] is None or r['no_live_reservation'][1] >= r['no_live_reservation'][0]) and (r['group_head'][0] == min(r['group_head'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'visible_now': [10, 10], 'not_expired': [9, 10], 'no_live_reservation': [8, 8], 'consumer_credit': [2, 3], 'group_head': [3, [3, 4]]}
check('valid operation', solve(r), True)
check('A scheduled message is delivered before its visibility instant', solve(dict(r, **{'visible_now': [9, 10]})), False)
check('A message is delivered after its expiration deadline', solve(dict(r, **{'not_expired': [10, 10]})), False)
check('A second worker reserves a currently leased message', solve(dict(r, **{'no_live_reservation': [9, 8]})), False)
check('The broker exceeds consumer in-flight credit', solve(dict(r, **{'consumer_credit': [3, 3]})), False)
check('A FIFO message overtakes an earlier group member', solve(dict(r, **{'group_head': [4, [3, 4]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'visible_now': [9, 10], 'not_expired': [10, 10], 'no_live_reservation': [9, 8], 'consumer_credit': [3, 3], 'group_head': [4, [3, 4]]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue reservation operation is admitted even though a second worker reserves a currently leased message.

Return a Boolean admission decision for reserve a queued message for delivery. The record r must satisfy all of: r['visible_now'][0] >= r['visible_now'][1]; r['not_expired'][0] < r['not_expired'][1]; r['no_live_reservation'][0] is None or r['no_live_reservation'][1] >= r['no_live_reservation'][0]; r['consumer_credit'][0] < r['consumer_credit'][1]; r['group_head'][0] == min(r['group_head'][1]). Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['visible_now'][0] >= r['visible_now'][1]) and (r['not_expired'][0] < r['not_expired'][1]) and (r['no_live_reservation'][0] is None or r['no_live_reservation'][1] >= r['no_live_reservation'][0]) and (r['group_head'][0] == min(r['group_head'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'visible_now': [10, 10], 'not_expired': [9, 10], 'no_live_reservation': [8, 8], 'consumer_credit': [2, 3], 'group_head': [3, [3, 4]]}
check('valid operation', solve(r), True)
check('A scheduled message is delivered before its visibility instant', solve(dict(r, **{'visible_now': [9, 10]})), False)
check('A message is delivered after its expiration deadline', solve(dict(r, **{'not_expired': [10, 10]})), False)
check('A second worker reserves a currently leased message', solve(dict(r, **{'no_live_reservation': [9, 8]})), False)
check('The broker exceeds consumer in-flight credit', solve(dict(r, **{'consumer_credit': [3, 3]})), False)
check('A FIFO message overtakes an earlier group member', solve(dict(r, **{'group_head': [4, [3, 4]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'visible_now': [9, 10], 'not_expired': [10, 10], 'no_live_reservation': [9, 8], 'consumer_credit': [3, 3], 'group_head': [4, [3, 4]]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

meaning: Fixture-count band, not measured model difficulty

deps: stdlib

evaluation_group: xd-queue-reservation

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['visible_now'][0] >= r['visible_now'][1]) and (r['not_expired'][0] < r['not_expired'][1]) and (r['no_live_reservation'][0] is None or r['no_live_reservation'][1] >= r['no_live_reservation'][0]) and (r['consumer_credit'][0] < r['consumer_credit'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'visible_now': [10, 10], 'not_expired': [9, 10], 'no_live_reservation': [8, 8], 'consumer_credit': [2, 3], 'group_head': [3, [3, 4]]}
check('valid operation', solve(r), True)
check('A scheduled message is delivered before its visibility instant', solve(dict(r, **{'visible_now': [9, 10]})), False)
check('A message is delivered after its expiration deadline', solve(dict(r, **{'not_expired': [10, 10]})), False)
check('A second worker reserves a currently leased message', solve(dict(r, **{'no_live_reservation': [9, 8]})), False)
check('The broker exceeds consumer in-flight credit', solve(dict(r, **{'consumer_credit': [3, 3]})), False)
check('A FIFO message overtakes an earlier group member', solve(dict(r, **{'group_head': [4, [3, 4]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'visible_now': [9, 10], 'not_expired': [10, 10], 'no_live_reservation': [9, 8], 'consumer_credit': [3, 3], 'group_head': [4, [3, 4]]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue reservation operation is admitted even though the broker exceeds consumer in-flight credit.

Return a Boolean admission decision for reserve a queued message for delivery. The record r must satisfy all of: r['visible_now'][0] >= r['visible_now'][1]; r['not_expired'][0] < r['not_expired'][1]; r['no_live_reservation'][0] is None or r['no_live_reservation'][1] >= r['no_live_reservation'][0]; r['consumer_credit'][0] < r['consumer_credit'][1]; r['group_head'][0] == min(r['group_head'][1]). Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['visible_now'][0] >= r['visible_now'][1]) and (r['not_expired'][0] < r['not_expired'][1]) and (r['no_live_reservation'][0] is None or r['no_live_reservation'][1] >= r['no_live_reservation'][0]) and (r['consumer_credit'][0] < r['consumer_credit'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'visible_now': [10, 10], 'not_expired': [9, 10], 'no_live_reservation': [8, 8], 'consumer_credit': [2, 3], 'group_head': [3, [3, 4]]}
check('valid operation', solve(r), True)
check('A scheduled message is delivered before its visibility instant', solve(dict(r, **{'visible_now': [9, 10]})), False)
check('A message is delivered after its expiration deadline', solve(dict(r, **{'not_expired': [10, 10]})), False)
check('A second worker reserves a currently leased message', solve(dict(r, **{'no_live_reservation': [9, 8]})), False)
check('The broker exceeds consumer in-flight credit', solve(dict(r, **{'consumer_credit': [3, 3]})), False)
check('A FIFO message overtakes an earlier group member', solve(dict(r, **{'group_head': [4, [3, 4]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'visible_now': [9, 10], 'not_expired': [10, 10], 'no_live_reservation': [9, 8], 'consumer_credit': [3, 3], 'group_head': [4, [3, 4]]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-reservation

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['not_expired'][0] < r['not_expired'][1]) and (r['no_live_reservation'][0] is None or r['no_live_reservation'][1] >= r['no_live_reservation'][0]) and (r['consumer_credit'][0] < r['consumer_credit'][1]) and (r['group_head'][0] == min(r['group_head'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'visible_now': [10, 10], 'not_expired': [9, 10], 'no_live_reservation': [8, 8], 'consumer_credit': [2, 3], 'group_head': [3, [3, 4]]}
check('valid operation', solve(r), True)
check('A scheduled message is delivered before its visibility instant', solve(dict(r, **{'visible_now': [9, 10]})), False)
check('A message is delivered after its expiration deadline', solve(dict(r, **{'not_expired': [10, 10]})), False)
check('A second worker reserves a currently leased message', solve(dict(r, **{'no_live_reservation': [9, 8]})), False)
check('The broker exceeds consumer in-flight credit', solve(dict(r, **{'consumer_credit': [3, 3]})), False)
check('A FIFO message overtakes an earlier group member', solve(dict(r, **{'group_head': [4, [3, 4]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'visible_now': [9, 10], 'not_expired': [10, 10], 'no_live_reservation': [9, 8], 'consumer_credit': [3, 3], 'group_head': [4, [3, 4]]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue reservation operation is admitted even though a fifo message overtakes an earlier group member.

Return a Boolean admission decision for reserve a queued message for delivery. The record r must satisfy all of: r['visible_now'][0] >= r['visible_now'][1]; r['not_expired'][0] < r['not_expired'][1]; r['no_live_reservation'][0] is None or r['no_live_reservation'][1] >= r['no_live_reservation'][0]; r['consumer_credit'][0] < r['consumer_credit'][1]; r['group_head'][0] == min(r['group_head'][1]). Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['lease_live'][0] < r['lease_live'][1]) and (r['channel_scope'][0] == r['channel_scope'][1]) and (r['settlement_once'] is False) and (all(r['batch_prefix']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'receipt_handle': ['h2', 'h2'], 'lease_live': [8, 9], 'channel_scope': ['c2', 'c2'], 'settlement_once': False, 'batch_prefix': [True, True, True]}
check('valid operation', solve(r), True)
check('A receipt from an older delivery deletes the current message', solve(dict(r, **{'receipt_handle': ['h1', 'h2']})), False)
check('An acknowledgement settles an already expired reservation', solve(dict(r, **{'lease_live': [9, 9]})), False)
check('A delivery tag from another channel is settled', solve(dict(r, **{'channel_scope': ['c1', 'c2']})), False)
check('A second settlement credits the consumer twice', solve(dict(r, **{'settlement_once': True})), False)
check('Cumulative acknowledgement jumps over an unprocessed delivery', solve(dict(r, **{'batch_prefix': [True, False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'receipt_handle': ['h1', 'h2'], 'lease_live': [9, 9], 'channel_scope': ['c1', 'c2'], 'settlement_once': True, 'batch_prefix': [True, False, True]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-acknowledge

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['receipt_handle'][0] == r['receipt_handle'][1]) and (r['channel_scope'][0] == r['channel_scope'][1]) and (r['settlement_once'] is False) and (all(r['batch_prefix']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'receipt_handle': ['h2', 'h2'], 'lease_live': [8, 9], 'channel_scope': ['c2', 'c2'], 'settlement_once': False, 'batch_prefix': [True, True, True]}
check('valid operation', solve(r), True)
check('A receipt from an older delivery deletes the current message', solve(dict(r, **{'receipt_handle': ['h1', 'h2']})), False)
check('An acknowledgement settles an already expired reservation', solve(dict(r, **{'lease_live': [9, 9]})), False)
check('A delivery tag from another channel is settled', solve(dict(r, **{'channel_scope': ['c1', 'c2']})), False)
check('A second settlement credits the consumer twice', solve(dict(r, **{'settlement_once': True})), False)
check('Cumulative acknowledgement jumps over an unprocessed delivery', solve(dict(r, **{'batch_prefix': [True, False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'receipt_handle': ['h1', 'h2'], 'lease_live': [9, 9], 'channel_scope': ['c1', 'c2'], 'settlement_once': True, 'batch_prefix': [True, False, True]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue acknowledge operation is admitted even though a receipt from an older delivery deletes the current message.

Return a Boolean admission decision for settle a queue delivery acknowledgement. The record r must satisfy all of: r['receipt_handle'][0] == r['receipt_handle'][1]; r['lease_live'][0] < r['lease_live'][1]; r['channel_scope'][0] == r['channel_scope'][1]; r['settlement_once'] is False; all(r['batch_prefix']). Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['receipt_handle'][0] == r['receipt_handle'][1]) and (r['channel_scope'][0] == r['channel_scope'][1]) and (r['settlement_once'] is False) and (all(r['batch_prefix']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'receipt_handle': ['h2', 'h2'], 'lease_live': [8, 9], 'channel_scope': ['c2', 'c2'], 'settlement_once': False, 'batch_prefix': [True, True, True]}
check('valid operation', solve(r), True)
check('A receipt from an older delivery deletes the current message', solve(dict(r, **{'receipt_handle': ['h1', 'h2']})), False)
check('An acknowledgement settles an already expired reservation', solve(dict(r, **{'lease_live': [9, 9]})), False)
check('A delivery tag from another channel is settled', solve(dict(r, **{'channel_scope': ['c1', 'c2']})), False)
check('A second settlement credits the consumer twice', solve(dict(r, **{'settlement_once': True})), False)
check('Cumulative acknowledgement jumps over an unprocessed delivery', solve(dict(r, **{'batch_prefix': [True, False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'receipt_handle': ['h1', 'h2'], 'lease_live': [9, 9], 'channel_scope': ['c1', 'c2'], 'settlement_once': True, 'batch_prefix': [True, False, True]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-acknowledge

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['receipt_handle'][0] == r['receipt_handle'][1]) and (r['lease_live'][0] < r['lease_live'][1]) and (r['settlement_once'] is False) and (all(r['batch_prefix']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'receipt_handle': ['h2', 'h2'], 'lease_live': [8, 9], 'channel_scope': ['c2', 'c2'], 'settlement_once': False, 'batch_prefix': [True, True, True]}
check('valid operation', solve(r), True)
check('A receipt from an older delivery deletes the current message', solve(dict(r, **{'receipt_handle': ['h1', 'h2']})), False)
check('An acknowledgement settles an already expired reservation', solve(dict(r, **{'lease_live': [9, 9]})), False)
check('A delivery tag from another channel is settled', solve(dict(r, **{'channel_scope': ['c1', 'c2']})), False)
check('A second settlement credits the consumer twice', solve(dict(r, **{'settlement_once': True})), False)
check('Cumulative acknowledgement jumps over an unprocessed delivery', solve(dict(r, **{'batch_prefix': [True, False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'receipt_handle': ['h1', 'h2'], 'lease_live': [9, 9], 'channel_scope': ['c1', 'c2'], 'settlement_once': True, 'batch_prefix': [True, False, True]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue acknowledge operation is admitted even though an acknowledgement settles an already expired reservation.

Return a Boolean admission decision for settle a queue delivery acknowledgement. The record r must satisfy all of: r['receipt_handle'][0] == r['receipt_handle'][1]; r['lease_live'][0] < r['lease_live'][1]; r['channel_scope'][0] == r['channel_scope'][1]; r['settlement_once'] is False; all(r['batch_prefix']). Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['receipt_handle'][0] == r['receipt_handle'][1]) and (r['lease_live'][0] < r['lease_live'][1]) and (r['settlement_once'] is False) and (all(r['batch_prefix']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'receipt_handle': ['h2', 'h2'], 'lease_live': [8, 9], 'channel_scope': ['c2', 'c2'], 'settlement_once': False, 'batch_prefix': [True, True, True]}
check('valid operation', solve(r), True)
check('A receipt from an older delivery deletes the current message', solve(dict(r, **{'receipt_handle': ['h1', 'h2']})), False)
check('An acknowledgement settles an already expired reservation', solve(dict(r, **{'lease_live': [9, 9]})), False)
check('A delivery tag from another channel is settled', solve(dict(r, **{'channel_scope': ['c1', 'c2']})), False)
check('A second settlement credits the consumer twice', solve(dict(r, **{'settlement_once': True})), False)
check('Cumulative acknowledgement jumps over an unprocessed delivery', solve(dict(r, **{'batch_prefix': [True, False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'receipt_handle': ['h1', 'h2'], 'lease_live': [9, 9], 'channel_scope': ['c1', 'c2'], 'settlement_once': True, 'batch_prefix': [True, False, True]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-acknowledge

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['receipt_handle'][0] == r['receipt_handle'][1]) and (r['lease_live'][0] < r['lease_live'][1]) and (r['channel_scope'][0] == r['channel_scope'][1]) and (all(r['batch_prefix']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'receipt_handle': ['h2', 'h2'], 'lease_live': [8, 9], 'channel_scope': ['c2', 'c2'], 'settlement_once': False, 'batch_prefix': [True, True, True]}
check('valid operation', solve(r), True)
check('A receipt from an older delivery deletes the current message', solve(dict(r, **{'receipt_handle': ['h1', 'h2']})), False)
check('An acknowledgement settles an already expired reservation', solve(dict(r, **{'lease_live': [9, 9]})), False)
check('A delivery tag from another channel is settled', solve(dict(r, **{'channel_scope': ['c1', 'c2']})), False)
check('A second settlement credits the consumer twice', solve(dict(r, **{'settlement_once': True})), False)
check('Cumulative acknowledgement jumps over an unprocessed delivery', solve(dict(r, **{'batch_prefix': [True, False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'receipt_handle': ['h1', 'h2'], 'lease_live': [9, 9], 'channel_scope': ['c1', 'c2'], 'settlement_once': True, 'batch_prefix': [True, False, True]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue acknowledge operation is admitted even though a delivery tag from another channel is settled.

Return a Boolean admission decision for settle a queue delivery acknowledgement. The record r must satisfy all of: r['receipt_handle'][0] == r['receipt_handle'][1]; r['lease_live'][0] < r['lease_live'][1]; r['channel_scope'][0] == r['channel_scope'][1]; r['settlement_once'] is False; all(r['batch_prefix']). Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['receipt_handle'][0] == r['receipt_handle'][1]) and (r['lease_live'][0] < r['lease_live'][1]) and (r['channel_scope'][0] == r['channel_scope'][1]) and (all(r['batch_prefix']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'receipt_handle': ['h2', 'h2'], 'lease_live': [8, 9], 'channel_scope': ['c2', 'c2'], 'settlement_once': False, 'batch_prefix': [True, True, True]}
check('valid operation', solve(r), True)
check('A receipt from an older delivery deletes the current message', solve(dict(r, **{'receipt_handle': ['h1', 'h2']})), False)
check('An acknowledgement settles an already expired reservation', solve(dict(r, **{'lease_live': [9, 9]})), False)
check('A delivery tag from another channel is settled', solve(dict(r, **{'channel_scope': ['c1', 'c2']})), False)
check('A second settlement credits the consumer twice', solve(dict(r, **{'settlement_once': True})), False)
check('Cumulative acknowledgement jumps over an unprocessed delivery', solve(dict(r, **{'batch_prefix': [True, False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'receipt_handle': ['h1', 'h2'], 'lease_live': [9, 9], 'channel_scope': ['c1', 'c2'], 'settlement_once': True, 'batch_prefix': [True, False, True]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-acknowledge

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['receipt_handle'][0] == r['receipt_handle'][1]) and (r['lease_live'][0] < r['lease_live'][1]) and (r['channel_scope'][0] == r['channel_scope'][1]) and (r['settlement_once'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'receipt_handle': ['h2', 'h2'], 'lease_live': [8, 9], 'channel_scope': ['c2', 'c2'], 'settlement_once': False, 'batch_prefix': [True, True, True]}
check('valid operation', solve(r), True)
check('A receipt from an older delivery deletes the current message', solve(dict(r, **{'receipt_handle': ['h1', 'h2']})), False)
check('An acknowledgement settles an already expired reservation', solve(dict(r, **{'lease_live': [9, 9]})), False)
check('A delivery tag from another channel is settled', solve(dict(r, **{'channel_scope': ['c1', 'c2']})), False)
check('A second settlement credits the consumer twice', solve(dict(r, **{'settlement_once': True})), False)
check('Cumulative acknowledgement jumps over an unprocessed delivery', solve(dict(r, **{'batch_prefix': [True, False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'receipt_handle': ['h1', 'h2'], 'lease_live': [9, 9], 'channel_scope': ['c1', 'c2'], 'settlement_once': True, 'batch_prefix': [True, False, True]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue acknowledge operation is admitted even though a second settlement credits the consumer twice.

Return a Boolean admission decision for settle a queue delivery acknowledgement. The record r must satisfy all of: r['receipt_handle'][0] == r['receipt_handle'][1]; r['lease_live'][0] < r['lease_live'][1]; r['channel_scope'][0] == r['channel_scope'][1]; r['settlement_once'] is False; all(r['batch_prefix']). Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['receipt_handle'][0] == r['receipt_handle'][1]) and (r['lease_live'][0] < r['lease_live'][1]) and (r['channel_scope'][0] == r['channel_scope'][1]) and (r['settlement_once'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'receipt_handle': ['h2', 'h2'], 'lease_live': [8, 9], 'channel_scope': ['c2', 'c2'], 'settlement_once': False, 'batch_prefix': [True, True, True]}
check('valid operation', solve(r), True)
check('A receipt from an older delivery deletes the current message', solve(dict(r, **{'receipt_handle': ['h1', 'h2']})), False)
check('An acknowledgement settles an already expired reservation', solve(dict(r, **{'lease_live': [9, 9]})), False)
check('A delivery tag from another channel is settled', solve(dict(r, **{'channel_scope': ['c1', 'c2']})), False)
check('A second settlement credits the consumer twice', solve(dict(r, **{'settlement_once': True})), False)
check('Cumulative acknowledgement jumps over an unprocessed delivery', solve(dict(r, **{'batch_prefix': [True, False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'receipt_handle': ['h1', 'h2'], 'lease_live': [9, 9], 'channel_scope': ['c1', 'c2'], 'settlement_once': True, 'batch_prefix': [True, False, True]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-acknowledge

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['lease_live'][0] < r['lease_live'][1]) and (r['channel_scope'][0] == r['channel_scope'][1]) and (r['settlement_once'] is False) and (all(r['batch_prefix']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'receipt_handle': ['h2', 'h2'], 'lease_live': [8, 9], 'channel_scope': ['c2', 'c2'], 'settlement_once': False, 'batch_prefix': [True, True, True]}
check('valid operation', solve(r), True)
check('A receipt from an older delivery deletes the current message', solve(dict(r, **{'receipt_handle': ['h1', 'h2']})), False)
check('An acknowledgement settles an already expired reservation', solve(dict(r, **{'lease_live': [9, 9]})), False)
check('A delivery tag from another channel is settled', solve(dict(r, **{'channel_scope': ['c1', 'c2']})), False)
check('A second settlement credits the consumer twice', solve(dict(r, **{'settlement_once': True})), False)
check('Cumulative acknowledgement jumps over an unprocessed delivery', solve(dict(r, **{'batch_prefix': [True, False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'receipt_handle': ['h1', 'h2'], 'lease_live': [9, 9], 'channel_scope': ['c1', 'c2'], 'settlement_once': True, 'batch_prefix': [True, False, True]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue acknowledge operation is admitted even though cumulative acknowledgement jumps over an unprocessed delivery.

Return a Boolean admission decision for settle a queue delivery acknowledgement. The record r must satisfy all of: r['receipt_handle'][0] == r['receipt_handle'][1]; r['lease_live'][0] < r['lease_live'][1]; r['channel_scope'][0] == r['channel_scope'][1]; r['settlement_once'] is False; all(r['batch_prefix']). Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['destination_exists'] is True) and (r['transfer_persisted'] is True) and (r['cycle_detection'][0] not in r['cycle_detection'][1]) and (r['original_identity'][0] == r['original_identity'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'attempt_threshold': [5, 5], 'destination_exists': True, 'transfer_persisted': True, 'cycle_detection': ['q3', ['q1', 'q2']], 'original_identity': ['m1', 'm1']}
check('valid operation', solve(r), True)
check('A transient first failure is dead-lettered prematurely', solve(dict(r, **{'attempt_threshold': [4, 5]})), False)
check('The source is deleted without a dead-letter destination', solve(dict(r, **{'destination_exists': False})), False)
check('The source is removed before the target copy is durable', solve(dict(r, **{'transfer_persisted': False})), False)
check('Dead-letter forwarding revisits a queue in its own route', solve(dict(r, **{'cycle_detection': ['q1', ['q1', 'q2']]})), False)
check('Dead-letter transfer loses the original message identity', solve(dict(r, **{'original_identity': ['m2', 'm1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'attempt_threshold': [4, 5], 'destination_exists': False, 'transfer_persisted': False, 'cycle_detection': ['q1', ['q1', 'q2']], 'original_identity': ['m2', 'm1']}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-dead-letter

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['attempt_threshold'][0] >= r['attempt_threshold'][1]) and (r['transfer_persisted'] is True) and (r['cycle_detection'][0] not in r['cycle_detection'][1]) and (r['original_identity'][0] == r['original_identity'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'attempt_threshold': [5, 5], 'destination_exists': True, 'transfer_persisted': True, 'cycle_detection': ['q3', ['q1', 'q2']], 'original_identity': ['m1', 'm1']}
check('valid operation', solve(r), True)
check('A transient first failure is dead-lettered prematurely', solve(dict(r, **{'attempt_threshold': [4, 5]})), False)
check('The source is deleted without a dead-letter destination', solve(dict(r, **{'destination_exists': False})), False)
check('The source is removed before the target copy is durable', solve(dict(r, **{'transfer_persisted': False})), False)
check('Dead-letter forwarding revisits a queue in its own route', solve(dict(r, **{'cycle_detection': ['q1', ['q1', 'q2']]})), False)
check('Dead-letter transfer loses the original message identity', solve(dict(r, **{'original_identity': ['m2', 'm1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'attempt_threshold': [4, 5], 'destination_exists': False, 'transfer_persisted': False, 'cycle_detection': ['q1', ['q1', 'q2']], 'original_identity': ['m2', 'm1']}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue dead letter operation is admitted even though a transient first failure is dead-lettered prematurely.

Return a Boolean admission decision for move a poison message to a dead-letter queue. The record r must satisfy all of: r['attempt_threshold'][0] >= r['attempt_threshold'][1]; r['destination_exists'] is True; r['transfer_persisted'] is True; r['cycle_detection'][0] not in r['cycle_detection'][1]; r['original_identity'][0] == r['original_identity'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['attempt_threshold'][0] >= r['attempt_threshold'][1]) and (r['transfer_persisted'] is True) and (r['cycle_detection'][0] not in r['cycle_detection'][1]) and (r['original_identity'][0] == r['original_identity'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'attempt_threshold': [5, 5], 'destination_exists': True, 'transfer_persisted': True, 'cycle_detection': ['q3', ['q1', 'q2']], 'original_identity': ['m1', 'm1']}
check('valid operation', solve(r), True)
check('A transient first failure is dead-lettered prematurely', solve(dict(r, **{'attempt_threshold': [4, 5]})), False)
check('The source is deleted without a dead-letter destination', solve(dict(r, **{'destination_exists': False})), False)
check('The source is removed before the target copy is durable', solve(dict(r, **{'transfer_persisted': False})), False)
check('Dead-letter forwarding revisits a queue in its own route', solve(dict(r, **{'cycle_detection': ['q1', ['q1', 'q2']]})), False)
check('Dead-letter transfer loses the original message identity', solve(dict(r, **{'original_identity': ['m2', 'm1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'attempt_threshold': [4, 5], 'destination_exists': False, 'transfer_persisted': False, 'cycle_detection': ['q1', ['q1', 'q2']], 'original_identity': ['m2', 'm1']}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-dead-letter

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['attempt_threshold'][0] >= r['attempt_threshold'][1]) and (r['destination_exists'] is True) and (r['cycle_detection'][0] not in r['cycle_detection'][1]) and (r['original_identity'][0] == r['original_identity'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'attempt_threshold': [5, 5], 'destination_exists': True, 'transfer_persisted': True, 'cycle_detection': ['q3', ['q1', 'q2']], 'original_identity': ['m1', 'm1']}
check('valid operation', solve(r), True)
check('A transient first failure is dead-lettered prematurely', solve(dict(r, **{'attempt_threshold': [4, 5]})), False)
check('The source is deleted without a dead-letter destination', solve(dict(r, **{'destination_exists': False})), False)
check('The source is removed before the target copy is durable', solve(dict(r, **{'transfer_persisted': False})), False)
check('Dead-letter forwarding revisits a queue in its own route', solve(dict(r, **{'cycle_detection': ['q1', ['q1', 'q2']]})), False)
check('Dead-letter transfer loses the original message identity', solve(dict(r, **{'original_identity': ['m2', 'm1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'attempt_threshold': [4, 5], 'destination_exists': False, 'transfer_persisted': False, 'cycle_detection': ['q1', ['q1', 'q2']], 'original_identity': ['m2', 'm1']}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue dead letter operation is admitted even though the source is deleted without a dead-letter destination.

Return a Boolean admission decision for move a poison message to a dead-letter queue. The record r must satisfy all of: r['attempt_threshold'][0] >= r['attempt_threshold'][1]; r['destination_exists'] is True; r['transfer_persisted'] is True; r['cycle_detection'][0] not in r['cycle_detection'][1]; r['original_identity'][0] == r['original_identity'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['attempt_threshold'][0] >= r['attempt_threshold'][1]) and (r['destination_exists'] is True) and (r['cycle_detection'][0] not in r['cycle_detection'][1]) and (r['original_identity'][0] == r['original_identity'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'attempt_threshold': [5, 5], 'destination_exists': True, 'transfer_persisted': True, 'cycle_detection': ['q3', ['q1', 'q2']], 'original_identity': ['m1', 'm1']}
check('valid operation', solve(r), True)
check('A transient first failure is dead-lettered prematurely', solve(dict(r, **{'attempt_threshold': [4, 5]})), False)
check('The source is deleted without a dead-letter destination', solve(dict(r, **{'destination_exists': False})), False)
check('The source is removed before the target copy is durable', solve(dict(r, **{'transfer_persisted': False})), False)
check('Dead-letter forwarding revisits a queue in its own route', solve(dict(r, **{'cycle_detection': ['q1', ['q1', 'q2']]})), False)
check('Dead-letter transfer loses the original message identity', solve(dict(r, **{'original_identity': ['m2', 'm1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'attempt_threshold': [4, 5], 'destination_exists': False, 'transfer_persisted': False, 'cycle_detection': ['q1', ['q1', 'q2']], 'original_identity': ['m2', 'm1']}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-dead-letter

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['attempt_threshold'][0] >= r['attempt_threshold'][1]) and (r['destination_exists'] is True) and (r['transfer_persisted'] is True) and (r['original_identity'][0] == r['original_identity'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'attempt_threshold': [5, 5], 'destination_exists': True, 'transfer_persisted': True, 'cycle_detection': ['q3', ['q1', 'q2']], 'original_identity': ['m1', 'm1']}
check('valid operation', solve(r), True)
check('A transient first failure is dead-lettered prematurely', solve(dict(r, **{'attempt_threshold': [4, 5]})), False)
check('The source is deleted without a dead-letter destination', solve(dict(r, **{'destination_exists': False})), False)
check('The source is removed before the target copy is durable', solve(dict(r, **{'transfer_persisted': False})), False)
check('Dead-letter forwarding revisits a queue in its own route', solve(dict(r, **{'cycle_detection': ['q1', ['q1', 'q2']]})), False)
check('Dead-letter transfer loses the original message identity', solve(dict(r, **{'original_identity': ['m2', 'm1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'attempt_threshold': [4, 5], 'destination_exists': False, 'transfer_persisted': False, 'cycle_detection': ['q1', ['q1', 'q2']], 'original_identity': ['m2', 'm1']}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue dead letter operation is admitted even though the source is removed before the target copy is durable.

Return a Boolean admission decision for move a poison message to a dead-letter queue. The record r must satisfy all of: r['attempt_threshold'][0] >= r['attempt_threshold'][1]; r['destination_exists'] is True; r['transfer_persisted'] is True; r['cycle_detection'][0] not in r['cycle_detection'][1]; r['original_identity'][0] == r['original_identity'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['attempt_threshold'][0] >= r['attempt_threshold'][1]) and (r['destination_exists'] is True) and (r['transfer_persisted'] is True) and (r['original_identity'][0] == r['original_identity'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'attempt_threshold': [5, 5], 'destination_exists': True, 'transfer_persisted': True, 'cycle_detection': ['q3', ['q1', 'q2']], 'original_identity': ['m1', 'm1']}
check('valid operation', solve(r), True)
check('A transient first failure is dead-lettered prematurely', solve(dict(r, **{'attempt_threshold': [4, 5]})), False)
check('The source is deleted without a dead-letter destination', solve(dict(r, **{'destination_exists': False})), False)
check('The source is removed before the target copy is durable', solve(dict(r, **{'transfer_persisted': False})), False)
check('Dead-letter forwarding revisits a queue in its own route', solve(dict(r, **{'cycle_detection': ['q1', ['q1', 'q2']]})), False)
check('Dead-letter transfer loses the original message identity', solve(dict(r, **{'original_identity': ['m2', 'm1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'attempt_threshold': [4, 5], 'destination_exists': False, 'transfer_persisted': False, 'cycle_detection': ['q1', ['q1', 'q2']], 'original_identity': ['m2', 'm1']}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-dead-letter

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['attempt_threshold'][0] >= r['attempt_threshold'][1]) and (r['destination_exists'] is True) and (r['transfer_persisted'] is True) and (r['cycle_detection'][0] not in r['cycle_detection'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'attempt_threshold': [5, 5], 'destination_exists': True, 'transfer_persisted': True, 'cycle_detection': ['q3', ['q1', 'q2']], 'original_identity': ['m1', 'm1']}
check('valid operation', solve(r), True)
check('A transient first failure is dead-lettered prematurely', solve(dict(r, **{'attempt_threshold': [4, 5]})), False)
check('The source is deleted without a dead-letter destination', solve(dict(r, **{'destination_exists': False})), False)
check('The source is removed before the target copy is durable', solve(dict(r, **{'transfer_persisted': False})), False)
check('Dead-letter forwarding revisits a queue in its own route', solve(dict(r, **{'cycle_detection': ['q1', ['q1', 'q2']]})), False)
check('Dead-letter transfer loses the original message identity', solve(dict(r, **{'original_identity': ['m2', 'm1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'attempt_threshold': [4, 5], 'destination_exists': False, 'transfer_persisted': False, 'cycle_detection': ['q1', ['q1', 'q2']], 'original_identity': ['m2', 'm1']}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue dead letter operation is admitted even though dead-letter forwarding revisits a queue in its own route.

Return a Boolean admission decision for move a poison message to a dead-letter queue. The record r must satisfy all of: r['attempt_threshold'][0] >= r['attempt_threshold'][1]; r['destination_exists'] is True; r['transfer_persisted'] is True; r['cycle_detection'][0] not in r['cycle_detection'][1]; r['original_identity'][0] == r['original_identity'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['attempt_threshold'][0] >= r['attempt_threshold'][1]) and (r['destination_exists'] is True) and (r['transfer_persisted'] is True) and (r['cycle_detection'][0] not in r['cycle_detection'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'attempt_threshold': [5, 5], 'destination_exists': True, 'transfer_persisted': True, 'cycle_detection': ['q3', ['q1', 'q2']], 'original_identity': ['m1', 'm1']}
check('valid operation', solve(r), True)
check('A transient first failure is dead-lettered prematurely', solve(dict(r, **{'attempt_threshold': [4, 5]})), False)
check('The source is deleted without a dead-letter destination', solve(dict(r, **{'destination_exists': False})), False)
check('The source is removed before the target copy is durable', solve(dict(r, **{'transfer_persisted': False})), False)
check('Dead-letter forwarding revisits a queue in its own route', solve(dict(r, **{'cycle_detection': ['q1', ['q1', 'q2']]})), False)
check('Dead-letter transfer loses the original message identity', solve(dict(r, **{'original_identity': ['m2', 'm1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'attempt_threshold': [4, 5], 'destination_exists': False, 'transfer_persisted': False, 'cycle_detection': ['q1', ['q1', 'q2']], 'original_identity': ['m2', 'm1']}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-dead-letter

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['destination_exists'] is True) and (r['transfer_persisted'] is True) and (r['cycle_detection'][0] not in r['cycle_detection'][1]) and (r['original_identity'][0] == r['original_identity'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'attempt_threshold': [5, 5], 'destination_exists': True, 'transfer_persisted': True, 'cycle_detection': ['q3', ['q1', 'q2']], 'original_identity': ['m1', 'm1']}
check('valid operation', solve(r), True)
check('A transient first failure is dead-lettered prematurely', solve(dict(r, **{'attempt_threshold': [4, 5]})), False)
check('The source is deleted without a dead-letter destination', solve(dict(r, **{'destination_exists': False})), False)
check('The source is removed before the target copy is durable', solve(dict(r, **{'transfer_persisted': False})), False)
check('Dead-letter forwarding revisits a queue in its own route', solve(dict(r, **{'cycle_detection': ['q1', ['q1', 'q2']]})), False)
check('Dead-letter transfer loses the original message identity', solve(dict(r, **{'original_identity': ['m2', 'm1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'attempt_threshold': [4, 5], 'destination_exists': False, 'transfer_persisted': False, 'cycle_detection': ['q1', ['q1', 'q2']], 'original_identity': ['m2', 'm1']}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue dead letter operation is admitted even though dead-letter transfer loses the original message identity.

Return a Boolean admission decision for move a poison message to a dead-letter queue. The record r must satisfy all of: r['attempt_threshold'][0] >= r['attempt_threshold'][1]; r['destination_exists'] is True; r['transfer_persisted'] is True; r['cycle_detection'][0] not in r['cycle_detection'][1]; r['original_identity'][0] == r['original_identity'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['source_version'][0] == r['source_version'][1]) and (r['destination_schema'][0] in r['destination_schema'][1]) and (r['retention_remaining'][0] < r['retention_remaining'][1]) and (r['rate_tokens'][0] >= r['rate_tokens'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'redrive_budget': [2, 3], 'source_version': [7, 7], 'destination_schema': [2, [2, 3]], 'retention_remaining': [19, 20], 'rate_tokens': [3, 3]}
check('valid operation', solve(r), True)
check('Poison messages circulate past the redrive allowance', solve(dict(r, **{'redrive_budget': [3, 3]})), False)
check('Redrive races an updated dead-letter record', solve(dict(r, **{'source_version': [6, 7]})), False)
check('The target consumer cannot decode the original schema', solve(dict(r, **{'destination_schema': [1, [2, 3]]})), False)
check('Redrive revives a message past its absolute retention', solve(dict(r, **{'retention_remaining': [20, 20]})), False)
check('Redrive bypasses its recovery traffic budget', solve(dict(r, **{'rate_tokens': [2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'redrive_budget': [3, 3], 'source_version': [6, 7], 'destination_schema': [1, [2, 3]], 'retention_remaining': [20, 20], 'rate_tokens': [2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-redrive

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['redrive_budget'][0] < r['redrive_budget'][1]) and (r['destination_schema'][0] in r['destination_schema'][1]) and (r['retention_remaining'][0] < r['retention_remaining'][1]) and (r['rate_tokens'][0] >= r['rate_tokens'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'redrive_budget': [2, 3], 'source_version': [7, 7], 'destination_schema': [2, [2, 3]], 'retention_remaining': [19, 20], 'rate_tokens': [3, 3]}
check('valid operation', solve(r), True)
check('Poison messages circulate past the redrive allowance', solve(dict(r, **{'redrive_budget': [3, 3]})), False)
check('Redrive races an updated dead-letter record', solve(dict(r, **{'source_version': [6, 7]})), False)
check('The target consumer cannot decode the original schema', solve(dict(r, **{'destination_schema': [1, [2, 3]]})), False)
check('Redrive revives a message past its absolute retention', solve(dict(r, **{'retention_remaining': [20, 20]})), False)
check('Redrive bypasses its recovery traffic budget', solve(dict(r, **{'rate_tokens': [2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'redrive_budget': [3, 3], 'source_version': [6, 7], 'destination_schema': [1, [2, 3]], 'retention_remaining': [20, 20], 'rate_tokens': [2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue redrive operation is admitted even though poison messages circulate past the redrive allowance.

Return a Boolean admission decision for redrive a previously dead-lettered message. The record r must satisfy all of: r['redrive_budget'][0] < r['redrive_budget'][1]; r['source_version'][0] == r['source_version'][1]; r['destination_schema'][0] in r['destination_schema'][1]; r['retention_remaining'][0] < r['retention_remaining'][1]; r['rate_tokens'][0] >= r['rate_tokens'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['redrive_budget'][0] < r['redrive_budget'][1]) and (r['destination_schema'][0] in r['destination_schema'][1]) and (r['retention_remaining'][0] < r['retention_remaining'][1]) and (r['rate_tokens'][0] >= r['rate_tokens'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'redrive_budget': [2, 3], 'source_version': [7, 7], 'destination_schema': [2, [2, 3]], 'retention_remaining': [19, 20], 'rate_tokens': [3, 3]}
check('valid operation', solve(r), True)
check('Poison messages circulate past the redrive allowance', solve(dict(r, **{'redrive_budget': [3, 3]})), False)
check('Redrive races an updated dead-letter record', solve(dict(r, **{'source_version': [6, 7]})), False)
check('The target consumer cannot decode the original schema', solve(dict(r, **{'destination_schema': [1, [2, 3]]})), False)
check('Redrive revives a message past its absolute retention', solve(dict(r, **{'retention_remaining': [20, 20]})), False)
check('Redrive bypasses its recovery traffic budget', solve(dict(r, **{'rate_tokens': [2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'redrive_budget': [3, 3], 'source_version': [6, 7], 'destination_schema': [1, [2, 3]], 'retention_remaining': [20, 20], 'rate_tokens': [2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-redrive

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['redrive_budget'][0] < r['redrive_budget'][1]) and (r['source_version'][0] == r['source_version'][1]) and (r['retention_remaining'][0] < r['retention_remaining'][1]) and (r['rate_tokens'][0] >= r['rate_tokens'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'redrive_budget': [2, 3], 'source_version': [7, 7], 'destination_schema': [2, [2, 3]], 'retention_remaining': [19, 20], 'rate_tokens': [3, 3]}
check('valid operation', solve(r), True)
check('Poison messages circulate past the redrive allowance', solve(dict(r, **{'redrive_budget': [3, 3]})), False)
check('Redrive races an updated dead-letter record', solve(dict(r, **{'source_version': [6, 7]})), False)
check('The target consumer cannot decode the original schema', solve(dict(r, **{'destination_schema': [1, [2, 3]]})), False)
check('Redrive revives a message past its absolute retention', solve(dict(r, **{'retention_remaining': [20, 20]})), False)
check('Redrive bypasses its recovery traffic budget', solve(dict(r, **{'rate_tokens': [2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'redrive_budget': [3, 3], 'source_version': [6, 7], 'destination_schema': [1, [2, 3]], 'retention_remaining': [20, 20], 'rate_tokens': [2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue redrive operation is admitted even though redrive races an updated dead-letter record.

Return a Boolean admission decision for redrive a previously dead-lettered message. The record r must satisfy all of: r['redrive_budget'][0] < r['redrive_budget'][1]; r['source_version'][0] == r['source_version'][1]; r['destination_schema'][0] in r['destination_schema'][1]; r['retention_remaining'][0] < r['retention_remaining'][1]; r['rate_tokens'][0] >= r['rate_tokens'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['redrive_budget'][0] < r['redrive_budget'][1]) and (r['source_version'][0] == r['source_version'][1]) and (r['retention_remaining'][0] < r['retention_remaining'][1]) and (r['rate_tokens'][0] >= r['rate_tokens'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'redrive_budget': [2, 3], 'source_version': [7, 7], 'destination_schema': [2, [2, 3]], 'retention_remaining': [19, 20], 'rate_tokens': [3, 3]}
check('valid operation', solve(r), True)
check('Poison messages circulate past the redrive allowance', solve(dict(r, **{'redrive_budget': [3, 3]})), False)
check('Redrive races an updated dead-letter record', solve(dict(r, **{'source_version': [6, 7]})), False)
check('The target consumer cannot decode the original schema', solve(dict(r, **{'destination_schema': [1, [2, 3]]})), False)
check('Redrive revives a message past its absolute retention', solve(dict(r, **{'retention_remaining': [20, 20]})), False)
check('Redrive bypasses its recovery traffic budget', solve(dict(r, **{'rate_tokens': [2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'redrive_budget': [3, 3], 'source_version': [6, 7], 'destination_schema': [1, [2, 3]], 'retention_remaining': [20, 20], 'rate_tokens': [2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-redrive

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['redrive_budget'][0] < r['redrive_budget'][1]) and (r['source_version'][0] == r['source_version'][1]) and (r['destination_schema'][0] in r['destination_schema'][1]) and (r['rate_tokens'][0] >= r['rate_tokens'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'redrive_budget': [2, 3], 'source_version': [7, 7], 'destination_schema': [2, [2, 3]], 'retention_remaining': [19, 20], 'rate_tokens': [3, 3]}
check('valid operation', solve(r), True)
check('Poison messages circulate past the redrive allowance', solve(dict(r, **{'redrive_budget': [3, 3]})), False)
check('Redrive races an updated dead-letter record', solve(dict(r, **{'source_version': [6, 7]})), False)
check('The target consumer cannot decode the original schema', solve(dict(r, **{'destination_schema': [1, [2, 3]]})), False)
check('Redrive revives a message past its absolute retention', solve(dict(r, **{'retention_remaining': [20, 20]})), False)
check('Redrive bypasses its recovery traffic budget', solve(dict(r, **{'rate_tokens': [2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'redrive_budget': [3, 3], 'source_version': [6, 7], 'destination_schema': [1, [2, 3]], 'retention_remaining': [20, 20], 'rate_tokens': [2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue redrive operation is admitted even though the target consumer cannot decode the original schema.

Return a Boolean admission decision for redrive a previously dead-lettered message. The record r must satisfy all of: r['redrive_budget'][0] < r['redrive_budget'][1]; r['source_version'][0] == r['source_version'][1]; r['destination_schema'][0] in r['destination_schema'][1]; r['retention_remaining'][0] < r['retention_remaining'][1]; r['rate_tokens'][0] >= r['rate_tokens'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['redrive_budget'][0] < r['redrive_budget'][1]) and (r['source_version'][0] == r['source_version'][1]) and (r['destination_schema'][0] in r['destination_schema'][1]) and (r['rate_tokens'][0] >= r['rate_tokens'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'redrive_budget': [2, 3], 'source_version': [7, 7], 'destination_schema': [2, [2, 3]], 'retention_remaining': [19, 20], 'rate_tokens': [3, 3]}
check('valid operation', solve(r), True)
check('Poison messages circulate past the redrive allowance', solve(dict(r, **{'redrive_budget': [3, 3]})), False)
check('Redrive races an updated dead-letter record', solve(dict(r, **{'source_version': [6, 7]})), False)
check('The target consumer cannot decode the original schema', solve(dict(r, **{'destination_schema': [1, [2, 3]]})), False)
check('Redrive revives a message past its absolute retention', solve(dict(r, **{'retention_remaining': [20, 20]})), False)
check('Redrive bypasses its recovery traffic budget', solve(dict(r, **{'rate_tokens': [2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'redrive_budget': [3, 3], 'source_version': [6, 7], 'destination_schema': [1, [2, 3]], 'retention_remaining': [20, 20], 'rate_tokens': [2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-redrive

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['redrive_budget'][0] < r['redrive_budget'][1]) and (r['source_version'][0] == r['source_version'][1]) and (r['destination_schema'][0] in r['destination_schema'][1]) and (r['retention_remaining'][0] < r['retention_remaining'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'redrive_budget': [2, 3], 'source_version': [7, 7], 'destination_schema': [2, [2, 3]], 'retention_remaining': [19, 20], 'rate_tokens': [3, 3]}
check('valid operation', solve(r), True)
check('Poison messages circulate past the redrive allowance', solve(dict(r, **{'redrive_budget': [3, 3]})), False)
check('Redrive races an updated dead-letter record', solve(dict(r, **{'source_version': [6, 7]})), False)
check('The target consumer cannot decode the original schema', solve(dict(r, **{'destination_schema': [1, [2, 3]]})), False)
check('Redrive revives a message past its absolute retention', solve(dict(r, **{'retention_remaining': [20, 20]})), False)
check('Redrive bypasses its recovery traffic budget', solve(dict(r, **{'rate_tokens': [2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'redrive_budget': [3, 3], 'source_version': [6, 7], 'destination_schema': [1, [2, 3]], 'retention_remaining': [20, 20], 'rate_tokens': [2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue redrive operation is admitted even though redrive revives a message past its absolute retention.

Return a Boolean admission decision for redrive a previously dead-lettered message. The record r must satisfy all of: r['redrive_budget'][0] < r['redrive_budget'][1]; r['source_version'][0] == r['source_version'][1]; r['destination_schema'][0] in r['destination_schema'][1]; r['retention_remaining'][0] < r['retention_remaining'][1]; r['rate_tokens'][0] >= r['rate_tokens'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['redrive_budget'][0] < r['redrive_budget'][1]) and (r['source_version'][0] == r['source_version'][1]) and (r['destination_schema'][0] in r['destination_schema'][1]) and (r['retention_remaining'][0] < r['retention_remaining'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'redrive_budget': [2, 3], 'source_version': [7, 7], 'destination_schema': [2, [2, 3]], 'retention_remaining': [19, 20], 'rate_tokens': [3, 3]}
check('valid operation', solve(r), True)
check('Poison messages circulate past the redrive allowance', solve(dict(r, **{'redrive_budget': [3, 3]})), False)
check('Redrive races an updated dead-letter record', solve(dict(r, **{'source_version': [6, 7]})), False)
check('The target consumer cannot decode the original schema', solve(dict(r, **{'destination_schema': [1, [2, 3]]})), False)
check('Redrive revives a message past its absolute retention', solve(dict(r, **{'retention_remaining': [20, 20]})), False)
check('Redrive bypasses its recovery traffic budget', solve(dict(r, **{'rate_tokens': [2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'redrive_budget': [3, 3], 'source_version': [6, 7], 'destination_schema': [1, [2, 3]], 'retention_remaining': [20, 20], 'rate_tokens': [2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-redrive

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['source_version'][0] == r['source_version'][1]) and (r['destination_schema'][0] in r['destination_schema'][1]) and (r['retention_remaining'][0] < r['retention_remaining'][1]) and (r['rate_tokens'][0] >= r['rate_tokens'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'redrive_budget': [2, 3], 'source_version': [7, 7], 'destination_schema': [2, [2, 3]], 'retention_remaining': [19, 20], 'rate_tokens': [3, 3]}
check('valid operation', solve(r), True)
check('Poison messages circulate past the redrive allowance', solve(dict(r, **{'redrive_budget': [3, 3]})), False)
check('Redrive races an updated dead-letter record', solve(dict(r, **{'source_version': [6, 7]})), False)
check('The target consumer cannot decode the original schema', solve(dict(r, **{'destination_schema': [1, [2, 3]]})), False)
check('Redrive revives a message past its absolute retention', solve(dict(r, **{'retention_remaining': [20, 20]})), False)
check('Redrive bypasses its recovery traffic budget', solve(dict(r, **{'rate_tokens': [2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'redrive_budget': [3, 3], 'source_version': [6, 7], 'destination_schema': [1, [2, 3]], 'retention_remaining': [20, 20], 'rate_tokens': [2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue redrive operation is admitted even though redrive bypasses its recovery traffic budget.

Return a Boolean admission decision for redrive a previously dead-lettered message. The record r must satisfy all of: r['redrive_budget'][0] < r['redrive_budget'][1]; r['source_version'][0] == r['source_version'][1]; r['destination_schema'][0] in r['destination_schema'][1]; r['retention_remaining'][0] < r['retention_remaining'][1]; r['rate_tokens'][0] >= r['rate_tokens'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['assignment_epoch'][0] > r['assignment_epoch'][1]) and (r['checkpoint_available'] is not None) and (r['subscription_match'][0] in r['subscription_match'][1]) and (r['capacity_limit'][0]+r['capacity_limit'][1] <= r['capacity_limit'][2])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'revocation_complete': True, 'assignment_epoch': [5, 4], 'checkpoint_available': 0, 'subscription_match': ['orders', ['orders', 'audit']], 'capacity_limit': [2, 1, 3]}
check('valid operation', solve(r), True)
check('The new consumer starts while the old owner remains active', solve(dict(r, **{'revocation_complete': False})), False)
check('A stale rebalance overwrites a newer assignment', solve(dict(r, **{'assignment_epoch': [4, 4]})), False)
check('A new owner starts without the last committed checkpoint', solve(dict(r, **{'checkpoint_available': None})), False)
check('A partition is assigned to a consumer not subscribed to its topic', solve(dict(r, **{'subscription_match': ['events', ['orders', 'audit']]})), False)
check('Rebalance exceeds the new owner partition capacity', solve(dict(r, **{'capacity_limit': [3, 1, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'revocation_complete': False, 'assignment_epoch': [4, 4], 'checkpoint_available': None, 'subscription_match': ['events', ['orders', 'audit']], 'capacity_limit': [3, 1, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-partition-rebalance

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['revocation_complete'] is True) and (r['checkpoint_available'] is not None) and (r['subscription_match'][0] in r['subscription_match'][1]) and (r['capacity_limit'][0]+r['capacity_limit'][1] <= r['capacity_limit'][2])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'revocation_complete': True, 'assignment_epoch': [5, 4], 'checkpoint_available': 0, 'subscription_match': ['orders', ['orders', 'audit']], 'capacity_limit': [2, 1, 3]}
check('valid operation', solve(r), True)
check('The new consumer starts while the old owner remains active', solve(dict(r, **{'revocation_complete': False})), False)
check('A stale rebalance overwrites a newer assignment', solve(dict(r, **{'assignment_epoch': [4, 4]})), False)
check('A new owner starts without the last committed checkpoint', solve(dict(r, **{'checkpoint_available': None})), False)
check('A partition is assigned to a consumer not subscribed to its topic', solve(dict(r, **{'subscription_match': ['events', ['orders', 'audit']]})), False)
check('Rebalance exceeds the new owner partition capacity', solve(dict(r, **{'capacity_limit': [3, 1, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'revocation_complete': False, 'assignment_epoch': [4, 4], 'checkpoint_available': None, 'subscription_match': ['events', ['orders', 'audit']], 'capacity_limit': [3, 1, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The partition rebalance operation is admitted even though the new consumer starts while the old owner remains active.

Return a Boolean admission decision for assign a partition during consumer rebalance. The record r must satisfy all of: r['revocation_complete'] is True; r['assignment_epoch'][0] > r['assignment_epoch'][1]; r['checkpoint_available'] is not None; r['subscription_match'][0] in r['subscription_match'][1]; r['capacity_limit'][0]+r['capacity_limit'][1] <= r['capacity_limit'][2]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['revocation_complete'] is True) and (r['checkpoint_available'] is not None) and (r['subscription_match'][0] in r['subscription_match'][1]) and (r['capacity_limit'][0]+r['capacity_limit'][1] <= r['capacity_limit'][2])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'revocation_complete': True, 'assignment_epoch': [5, 4], 'checkpoint_available': 0, 'subscription_match': ['orders', ['orders', 'audit']], 'capacity_limit': [2, 1, 3]}
check('valid operation', solve(r), True)
check('The new consumer starts while the old owner remains active', solve(dict(r, **{'revocation_complete': False})), False)
check('A stale rebalance overwrites a newer assignment', solve(dict(r, **{'assignment_epoch': [4, 4]})), False)
check('A new owner starts without the last committed checkpoint', solve(dict(r, **{'checkpoint_available': None})), False)
check('A partition is assigned to a consumer not subscribed to its topic', solve(dict(r, **{'subscription_match': ['events', ['orders', 'audit']]})), False)
check('Rebalance exceeds the new owner partition capacity', solve(dict(r, **{'capacity_limit': [3, 1, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'revocation_complete': False, 'assignment_epoch': [4, 4], 'checkpoint_available': None, 'subscription_match': ['events', ['orders', 'audit']], 'capacity_limit': [3, 1, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-partition-rebalance

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['revocation_complete'] is True) and (r['assignment_epoch'][0] > r['assignment_epoch'][1]) and (r['subscription_match'][0] in r['subscription_match'][1]) and (r['capacity_limit'][0]+r['capacity_limit'][1] <= r['capacity_limit'][2])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'revocation_complete': True, 'assignment_epoch': [5, 4], 'checkpoint_available': 0, 'subscription_match': ['orders', ['orders', 'audit']], 'capacity_limit': [2, 1, 3]}
check('valid operation', solve(r), True)
check('The new consumer starts while the old owner remains active', solve(dict(r, **{'revocation_complete': False})), False)
check('A stale rebalance overwrites a newer assignment', solve(dict(r, **{'assignment_epoch': [4, 4]})), False)
check('A new owner starts without the last committed checkpoint', solve(dict(r, **{'checkpoint_available': None})), False)
check('A partition is assigned to a consumer not subscribed to its topic', solve(dict(r, **{'subscription_match': ['events', ['orders', 'audit']]})), False)
check('Rebalance exceeds the new owner partition capacity', solve(dict(r, **{'capacity_limit': [3, 1, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'revocation_complete': False, 'assignment_epoch': [4, 4], 'checkpoint_available': None, 'subscription_match': ['events', ['orders', 'audit']], 'capacity_limit': [3, 1, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The partition rebalance operation is admitted even though a stale rebalance overwrites a newer assignment.

Return a Boolean admission decision for assign a partition during consumer rebalance. The record r must satisfy all of: r['revocation_complete'] is True; r['assignment_epoch'][0] > r['assignment_epoch'][1]; r['checkpoint_available'] is not None; r['subscription_match'][0] in r['subscription_match'][1]; r['capacity_limit'][0]+r['capacity_limit'][1] <= r['capacity_limit'][2]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['revocation_complete'] is True) and (r['assignment_epoch'][0] > r['assignment_epoch'][1]) and (r['subscription_match'][0] in r['subscription_match'][1]) and (r['capacity_limit'][0]+r['capacity_limit'][1] <= r['capacity_limit'][2])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'revocation_complete': True, 'assignment_epoch': [5, 4], 'checkpoint_available': 0, 'subscription_match': ['orders', ['orders', 'audit']], 'capacity_limit': [2, 1, 3]}
check('valid operation', solve(r), True)
check('The new consumer starts while the old owner remains active', solve(dict(r, **{'revocation_complete': False})), False)
check('A stale rebalance overwrites a newer assignment', solve(dict(r, **{'assignment_epoch': [4, 4]})), False)
check('A new owner starts without the last committed checkpoint', solve(dict(r, **{'checkpoint_available': None})), False)
check('A partition is assigned to a consumer not subscribed to its topic', solve(dict(r, **{'subscription_match': ['events', ['orders', 'audit']]})), False)
check('Rebalance exceeds the new owner partition capacity', solve(dict(r, **{'capacity_limit': [3, 1, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'revocation_complete': False, 'assignment_epoch': [4, 4], 'checkpoint_available': None, 'subscription_match': ['events', ['orders', 'audit']], 'capacity_limit': [3, 1, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-partition-rebalance

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(r):
    return (r['revocation_complete'] is True) and (r['assignment_epoch'][0] > r['assignment_epoch'][1]) and (r['checkpoint_available'] is not None) and (r['capacity_limit'][0]+r['capacity_limit'][1] <= r['capacity_limit'][2])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'revocation_complete': True, 'assignment_epoch': [5, 4], 'checkpoint_available': 0, 'subscription_match': ['orders', ['orders', 'audit']], 'capacity_limit': [2, 1, 3]}
check('valid operation', solve(r), True)
check('The new consumer starts while the old owner remains active', solve(dict(r, **{'revocation_complete': False})), False)
check('A stale rebalance overwrites a newer assignment', solve(dict(r, **{'assignment_epoch': [4, 4]})), False)
check('A new owner starts without the last committed checkpoint', solve(dict(r, **{'checkpoint_available': None})), False)
check('A partition is assigned to a consumer not subscribed to its topic', solve(dict(r, **{'subscription_match': ['events', ['orders', 'audit']]})), False)
check('Rebalance exceeds the new owner partition capacity', solve(dict(r, **{'capacity_limit': [3, 1, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'revocation_complete': False, 'assignment_epoch': [4, 4], 'checkpoint_available': None, 'subscription_match': ['events', ['orders', 'audit']], 'capacity_limit': [3, 1, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The partition rebalance operation is admitted even though a new owner starts without the last committed checkpoint.

Return a Boolean admission decision for assign a partition during consumer rebalance. The record r must satisfy all of: r['revocation_complete'] is True; r['assignment_epoch'][0] > r['assignment_epoch'][1]; r['checkpoint_available'] is not None; r['subscription_match'][0] in r['subscription_match'][1]; r['capacity_limit'][0]+r['capacity_limit'][1] <= r['capacity_limit'][2]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access