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

N = 1
observations = []
def solve(r):
    return (r['replacement_count'][0]-1 >= r['replacement_count'][1]) and (len(r['handoff_complete']) == 0) and (r['reader_drain'] == 0) and (r['membership_commit'][0] <= r['membership_commit'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'replacement_count': [4, 3], 'replacement_durable': True, 'handoff_complete': [], 'reader_drain': 0, 'membership_commit': [20, 20]}
check('valid operation', solve(r), True)
check('Retirement leaves fewer copies than the required factor', solve(dict(r, **{'replacement_count': [3, 3]})), False)
check('The replacement has only volatile copied state', solve(dict(r, **{'replacement_durable': False})), False)
check('Keys still assigned to the retiring replica are lost', solve(dict(r, **{'handoff_complete': ['key-7']})), False)
check('Active snapshot readers lose their pinned source', solve(dict(r, **{'reader_drain': 2})), False)
check('Physical removal precedes committed membership removal', solve(dict(r, **{'membership_commit': [21, 20]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'replacement_count': [3, 3], 'replacement_durable': False, 'handoff_complete': ['key-7'], 'reader_drain': 2, 'membership_commit': [21, 20]}
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-replica-removal

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

N = 1
observations = []
def solve(r):
    return (r['replacement_count'][0]-1 >= r['replacement_count'][1]) and (r['replacement_durable'] is True) and (r['reader_drain'] == 0) and (r['membership_commit'][0] <= r['membership_commit'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'replacement_count': [4, 3], 'replacement_durable': True, 'handoff_complete': [], 'reader_drain': 0, 'membership_commit': [20, 20]}
check('valid operation', solve(r), True)
check('Retirement leaves fewer copies than the required factor', solve(dict(r, **{'replacement_count': [3, 3]})), False)
check('The replacement has only volatile copied state', solve(dict(r, **{'replacement_durable': False})), False)
check('Keys still assigned to the retiring replica are lost', solve(dict(r, **{'handoff_complete': ['key-7']})), False)
check('Active snapshot readers lose their pinned source', solve(dict(r, **{'reader_drain': 2})), False)
check('Physical removal precedes committed membership removal', solve(dict(r, **{'membership_commit': [21, 20]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'replacement_count': [3, 3], 'replacement_durable': False, 'handoff_complete': ['key-7'], 'reader_drain': 2, 'membership_commit': [21, 20]}
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 replica removal operation is admitted even though the replacement has only volatile copied state.

Return a Boolean admission decision for retire a replica and discard its data. The record r must satisfy all of: r['replacement_count'][0]-1 >= r['replacement_count'][1]; r['replacement_durable'] is True; len(r['handoff_complete']) == 0; r['reader_drain'] == 0; r['membership_commit'][0] <= r['membership_commit'][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['replacement_count'][0]-1 >= r['replacement_count'][1]) and (r['replacement_durable'] is True) and (r['reader_drain'] == 0) and (r['membership_commit'][0] <= r['membership_commit'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'replacement_count': [4, 3], 'replacement_durable': True, 'handoff_complete': [], 'reader_drain': 0, 'membership_commit': [20, 20]}
check('valid operation', solve(r), True)
check('Retirement leaves fewer copies than the required factor', solve(dict(r, **{'replacement_count': [3, 3]})), False)
check('The replacement has only volatile copied state', solve(dict(r, **{'replacement_durable': False})), False)
check('Keys still assigned to the retiring replica are lost', solve(dict(r, **{'handoff_complete': ['key-7']})), False)
check('Active snapshot readers lose their pinned source', solve(dict(r, **{'reader_drain': 2})), False)
check('Physical removal precedes committed membership removal', solve(dict(r, **{'membership_commit': [21, 20]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'replacement_count': [3, 3], 'replacement_durable': False, 'handoff_complete': ['key-7'], 'reader_drain': 2, 'membership_commit': [21, 20]}
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-replica-removal

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

N = 1
observations = []
def solve(r):
    return (r['replacement_count'][0]-1 >= r['replacement_count'][1]) and (r['replacement_durable'] is True) and (len(r['handoff_complete']) == 0) and (r['membership_commit'][0] <= r['membership_commit'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'replacement_count': [4, 3], 'replacement_durable': True, 'handoff_complete': [], 'reader_drain': 0, 'membership_commit': [20, 20]}
check('valid operation', solve(r), True)
check('Retirement leaves fewer copies than the required factor', solve(dict(r, **{'replacement_count': [3, 3]})), False)
check('The replacement has only volatile copied state', solve(dict(r, **{'replacement_durable': False})), False)
check('Keys still assigned to the retiring replica are lost', solve(dict(r, **{'handoff_complete': ['key-7']})), False)
check('Active snapshot readers lose their pinned source', solve(dict(r, **{'reader_drain': 2})), False)
check('Physical removal precedes committed membership removal', solve(dict(r, **{'membership_commit': [21, 20]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'replacement_count': [3, 3], 'replacement_durable': False, 'handoff_complete': ['key-7'], 'reader_drain': 2, 'membership_commit': [21, 20]}
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 replica removal operation is admitted even though keys still assigned to the retiring replica are lost.

Return a Boolean admission decision for retire a replica and discard its data. The record r must satisfy all of: r['replacement_count'][0]-1 >= r['replacement_count'][1]; r['replacement_durable'] is True; len(r['handoff_complete']) == 0; r['reader_drain'] == 0; r['membership_commit'][0] <= r['membership_commit'][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['replacement_count'][0]-1 >= r['replacement_count'][1]) and (r['replacement_durable'] is True) and (len(r['handoff_complete']) == 0) and (r['membership_commit'][0] <= r['membership_commit'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'replacement_count': [4, 3], 'replacement_durable': True, 'handoff_complete': [], 'reader_drain': 0, 'membership_commit': [20, 20]}
check('valid operation', solve(r), True)
check('Retirement leaves fewer copies than the required factor', solve(dict(r, **{'replacement_count': [3, 3]})), False)
check('The replacement has only volatile copied state', solve(dict(r, **{'replacement_durable': False})), False)
check('Keys still assigned to the retiring replica are lost', solve(dict(r, **{'handoff_complete': ['key-7']})), False)
check('Active snapshot readers lose their pinned source', solve(dict(r, **{'reader_drain': 2})), False)
check('Physical removal precedes committed membership removal', solve(dict(r, **{'membership_commit': [21, 20]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'replacement_count': [3, 3], 'replacement_durable': False, 'handoff_complete': ['key-7'], 'reader_drain': 2, 'membership_commit': [21, 20]}
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-replica-removal

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

N = 1
observations = []
def solve(r):
    return (r['replacement_count'][0]-1 >= r['replacement_count'][1]) and (r['replacement_durable'] is True) and (len(r['handoff_complete']) == 0) and (r['reader_drain'] == 0)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'replacement_count': [4, 3], 'replacement_durable': True, 'handoff_complete': [], 'reader_drain': 0, 'membership_commit': [20, 20]}
check('valid operation', solve(r), True)
check('Retirement leaves fewer copies than the required factor', solve(dict(r, **{'replacement_count': [3, 3]})), False)
check('The replacement has only volatile copied state', solve(dict(r, **{'replacement_durable': False})), False)
check('Keys still assigned to the retiring replica are lost', solve(dict(r, **{'handoff_complete': ['key-7']})), False)
check('Active snapshot readers lose their pinned source', solve(dict(r, **{'reader_drain': 2})), False)
check('Physical removal precedes committed membership removal', solve(dict(r, **{'membership_commit': [21, 20]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'replacement_count': [3, 3], 'replacement_durable': False, 'handoff_complete': ['key-7'], 'reader_drain': 2, 'membership_commit': [21, 20]}
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 replica removal operation is admitted even though active snapshot readers lose their pinned source.

Return a Boolean admission decision for retire a replica and discard its data. The record r must satisfy all of: r['replacement_count'][0]-1 >= r['replacement_count'][1]; r['replacement_durable'] is True; len(r['handoff_complete']) == 0; r['reader_drain'] == 0; r['membership_commit'][0] <= r['membership_commit'][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['replacement_count'][0]-1 >= r['replacement_count'][1]) and (r['replacement_durable'] is True) and (len(r['handoff_complete']) == 0) and (r['reader_drain'] == 0)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'replacement_count': [4, 3], 'replacement_durable': True, 'handoff_complete': [], 'reader_drain': 0, 'membership_commit': [20, 20]}
check('valid operation', solve(r), True)
check('Retirement leaves fewer copies than the required factor', solve(dict(r, **{'replacement_count': [3, 3]})), False)
check('The replacement has only volatile copied state', solve(dict(r, **{'replacement_durable': False})), False)
check('Keys still assigned to the retiring replica are lost', solve(dict(r, **{'handoff_complete': ['key-7']})), False)
check('Active snapshot readers lose their pinned source', solve(dict(r, **{'reader_drain': 2})), False)
check('Physical removal precedes committed membership removal', solve(dict(r, **{'membership_commit': [21, 20]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'replacement_count': [3, 3], 'replacement_durable': False, 'handoff_complete': ['key-7'], 'reader_drain': 2, 'membership_commit': [21, 20]}
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-replica-removal

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

N = 1
observations = []
def solve(r):
    return (r['replacement_durable'] is True) and (len(r['handoff_complete']) == 0) and (r['reader_drain'] == 0) and (r['membership_commit'][0] <= r['membership_commit'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'replacement_count': [4, 3], 'replacement_durable': True, 'handoff_complete': [], 'reader_drain': 0, 'membership_commit': [20, 20]}
check('valid operation', solve(r), True)
check('Retirement leaves fewer copies than the required factor', solve(dict(r, **{'replacement_count': [3, 3]})), False)
check('The replacement has only volatile copied state', solve(dict(r, **{'replacement_durable': False})), False)
check('Keys still assigned to the retiring replica are lost', solve(dict(r, **{'handoff_complete': ['key-7']})), False)
check('Active snapshot readers lose their pinned source', solve(dict(r, **{'reader_drain': 2})), False)
check('Physical removal precedes committed membership removal', solve(dict(r, **{'membership_commit': [21, 20]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'replacement_count': [3, 3], 'replacement_durable': False, 'handoff_complete': ['key-7'], 'reader_drain': 2, 'membership_commit': [21, 20]}
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 replica removal operation is admitted even though physical removal precedes committed membership removal.

Return a Boolean admission decision for retire a replica and discard its data. The record r must satisfy all of: r['replacement_count'][0]-1 >= r['replacement_count'][1]; r['replacement_durable'] is True; len(r['handoff_complete']) == 0; r['reader_drain'] == 0; r['membership_commit'][0] <= r['membership_commit'][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['backup_pin'][0] not in r['backup_pin'][1]) and (r['archive_confirmed'] is True) and (r['segment_sealed'] is True) and (r['recovery_floor'][0] < r['recovery_floor'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'slowest_replica': [9, [10, 12]], 'backup_pin': [9, [10, 11]], 'archive_confirmed': True, 'segment_sealed': True, 'recovery_floor': [7, 8]}
check('valid operation', solve(r), True)
check('Deletion removes the next segment needed by a lagging replica', solve(dict(r, **{'slowest_replica': [10, [10, 12]]})), False)
check('A backup still needs the selected log segment', solve(dict(r, **{'backup_pin': [10, [10, 11]]})), False)
check('The only recoverable copy is deleted before archival', solve(dict(r, **{'archive_confirmed': False})), False)
check('Garbage collection deletes a segment still accepting writes', solve(dict(r, **{'segment_sealed': False})), False)
check('Deletion crosses the crash recovery replay floor', solve(dict(r, **{'recovery_floor': [8, 8]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'slowest_replica': [10, [10, 12]], 'backup_pin': [10, [10, 11]], 'archive_confirmed': False, 'segment_sealed': False, 'recovery_floor': [8, 8]}
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-wal-retention

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

N = 1
observations = []
def solve(r):
    return (r['slowest_replica'][0] < min(r['slowest_replica'][1])) and (r['archive_confirmed'] is True) and (r['segment_sealed'] is True) and (r['recovery_floor'][0] < r['recovery_floor'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'slowest_replica': [9, [10, 12]], 'backup_pin': [9, [10, 11]], 'archive_confirmed': True, 'segment_sealed': True, 'recovery_floor': [7, 8]}
check('valid operation', solve(r), True)
check('Deletion removes the next segment needed by a lagging replica', solve(dict(r, **{'slowest_replica': [10, [10, 12]]})), False)
check('A backup still needs the selected log segment', solve(dict(r, **{'backup_pin': [10, [10, 11]]})), False)
check('The only recoverable copy is deleted before archival', solve(dict(r, **{'archive_confirmed': False})), False)
check('Garbage collection deletes a segment still accepting writes', solve(dict(r, **{'segment_sealed': False})), False)
check('Deletion crosses the crash recovery replay floor', solve(dict(r, **{'recovery_floor': [8, 8]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'slowest_replica': [10, [10, 12]], 'backup_pin': [10, [10, 11]], 'archive_confirmed': False, 'segment_sealed': False, 'recovery_floor': [8, 8]}
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 wal retention operation is admitted even though deletion removes the next segment needed by a lagging replica.

Return a Boolean admission decision for delete a write-ahead log segment. The record r must satisfy all of: r['slowest_replica'][0] < min(r['slowest_replica'][1]); r['backup_pin'][0] not in r['backup_pin'][1]; r['archive_confirmed'] is True; r['segment_sealed'] is True; r['recovery_floor'][0] < r['recovery_floor'][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['slowest_replica'][0] < min(r['slowest_replica'][1])) and (r['archive_confirmed'] is True) and (r['segment_sealed'] is True) and (r['recovery_floor'][0] < r['recovery_floor'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'slowest_replica': [9, [10, 12]], 'backup_pin': [9, [10, 11]], 'archive_confirmed': True, 'segment_sealed': True, 'recovery_floor': [7, 8]}
check('valid operation', solve(r), True)
check('Deletion removes the next segment needed by a lagging replica', solve(dict(r, **{'slowest_replica': [10, [10, 12]]})), False)
check('A backup still needs the selected log segment', solve(dict(r, **{'backup_pin': [10, [10, 11]]})), False)
check('The only recoverable copy is deleted before archival', solve(dict(r, **{'archive_confirmed': False})), False)
check('Garbage collection deletes a segment still accepting writes', solve(dict(r, **{'segment_sealed': False})), False)
check('Deletion crosses the crash recovery replay floor', solve(dict(r, **{'recovery_floor': [8, 8]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'slowest_replica': [10, [10, 12]], 'backup_pin': [10, [10, 11]], 'archive_confirmed': False, 'segment_sealed': False, 'recovery_floor': [8, 8]}
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-wal-retention

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

N = 1
observations = []
def solve(r):
    return (r['slowest_replica'][0] < min(r['slowest_replica'][1])) and (r['backup_pin'][0] not in r['backup_pin'][1]) and (r['segment_sealed'] is True) and (r['recovery_floor'][0] < r['recovery_floor'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'slowest_replica': [9, [10, 12]], 'backup_pin': [9, [10, 11]], 'archive_confirmed': True, 'segment_sealed': True, 'recovery_floor': [7, 8]}
check('valid operation', solve(r), True)
check('Deletion removes the next segment needed by a lagging replica', solve(dict(r, **{'slowest_replica': [10, [10, 12]]})), False)
check('A backup still needs the selected log segment', solve(dict(r, **{'backup_pin': [10, [10, 11]]})), False)
check('The only recoverable copy is deleted before archival', solve(dict(r, **{'archive_confirmed': False})), False)
check('Garbage collection deletes a segment still accepting writes', solve(dict(r, **{'segment_sealed': False})), False)
check('Deletion crosses the crash recovery replay floor', solve(dict(r, **{'recovery_floor': [8, 8]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'slowest_replica': [10, [10, 12]], 'backup_pin': [10, [10, 11]], 'archive_confirmed': False, 'segment_sealed': False, 'recovery_floor': [8, 8]}
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 wal retention operation is admitted even though a backup still needs the selected log segment.

Return a Boolean admission decision for delete a write-ahead log segment. The record r must satisfy all of: r['slowest_replica'][0] < min(r['slowest_replica'][1]); r['backup_pin'][0] not in r['backup_pin'][1]; r['archive_confirmed'] is True; r['segment_sealed'] is True; r['recovery_floor'][0] < r['recovery_floor'][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['slowest_replica'][0] < min(r['slowest_replica'][1])) and (r['backup_pin'][0] not in r['backup_pin'][1]) and (r['segment_sealed'] is True) and (r['recovery_floor'][0] < r['recovery_floor'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'slowest_replica': [9, [10, 12]], 'backup_pin': [9, [10, 11]], 'archive_confirmed': True, 'segment_sealed': True, 'recovery_floor': [7, 8]}
check('valid operation', solve(r), True)
check('Deletion removes the next segment needed by a lagging replica', solve(dict(r, **{'slowest_replica': [10, [10, 12]]})), False)
check('A backup still needs the selected log segment', solve(dict(r, **{'backup_pin': [10, [10, 11]]})), False)
check('The only recoverable copy is deleted before archival', solve(dict(r, **{'archive_confirmed': False})), False)
check('Garbage collection deletes a segment still accepting writes', solve(dict(r, **{'segment_sealed': False})), False)
check('Deletion crosses the crash recovery replay floor', solve(dict(r, **{'recovery_floor': [8, 8]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'slowest_replica': [10, [10, 12]], 'backup_pin': [10, [10, 11]], 'archive_confirmed': False, 'segment_sealed': False, 'recovery_floor': [8, 8]}
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-wal-retention

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

N = 1
observations = []
def solve(r):
    return (r['slowest_replica'][0] < min(r['slowest_replica'][1])) and (r['backup_pin'][0] not in r['backup_pin'][1]) and (r['archive_confirmed'] is True) and (r['recovery_floor'][0] < r['recovery_floor'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'slowest_replica': [9, [10, 12]], 'backup_pin': [9, [10, 11]], 'archive_confirmed': True, 'segment_sealed': True, 'recovery_floor': [7, 8]}
check('valid operation', solve(r), True)
check('Deletion removes the next segment needed by a lagging replica', solve(dict(r, **{'slowest_replica': [10, [10, 12]]})), False)
check('A backup still needs the selected log segment', solve(dict(r, **{'backup_pin': [10, [10, 11]]})), False)
check('The only recoverable copy is deleted before archival', solve(dict(r, **{'archive_confirmed': False})), False)
check('Garbage collection deletes a segment still accepting writes', solve(dict(r, **{'segment_sealed': False})), False)
check('Deletion crosses the crash recovery replay floor', solve(dict(r, **{'recovery_floor': [8, 8]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'slowest_replica': [10, [10, 12]], 'backup_pin': [10, [10, 11]], 'archive_confirmed': False, 'segment_sealed': False, 'recovery_floor': [8, 8]}
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 wal retention operation is admitted even though the only recoverable copy is deleted before archival.

Return a Boolean admission decision for delete a write-ahead log segment. The record r must satisfy all of: r['slowest_replica'][0] < min(r['slowest_replica'][1]); r['backup_pin'][0] not in r['backup_pin'][1]; r['archive_confirmed'] is True; r['segment_sealed'] is True; r['recovery_floor'][0] < r['recovery_floor'][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['slowest_replica'][0] < min(r['slowest_replica'][1])) and (r['backup_pin'][0] not in r['backup_pin'][1]) and (r['archive_confirmed'] is True) and (r['recovery_floor'][0] < r['recovery_floor'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'slowest_replica': [9, [10, 12]], 'backup_pin': [9, [10, 11]], 'archive_confirmed': True, 'segment_sealed': True, 'recovery_floor': [7, 8]}
check('valid operation', solve(r), True)
check('Deletion removes the next segment needed by a lagging replica', solve(dict(r, **{'slowest_replica': [10, [10, 12]]})), False)
check('A backup still needs the selected log segment', solve(dict(r, **{'backup_pin': [10, [10, 11]]})), False)
check('The only recoverable copy is deleted before archival', solve(dict(r, **{'archive_confirmed': False})), False)
check('Garbage collection deletes a segment still accepting writes', solve(dict(r, **{'segment_sealed': False})), False)
check('Deletion crosses the crash recovery replay floor', solve(dict(r, **{'recovery_floor': [8, 8]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'slowest_replica': [10, [10, 12]], 'backup_pin': [10, [10, 11]], 'archive_confirmed': False, 'segment_sealed': False, 'recovery_floor': [8, 8]}
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-wal-retention

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

N = 1
observations = []
def solve(r):
    return (r['slowest_replica'][0] < min(r['slowest_replica'][1])) and (r['backup_pin'][0] not in r['backup_pin'][1]) and (r['archive_confirmed'] is True) and (r['segment_sealed'] is True)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'slowest_replica': [9, [10, 12]], 'backup_pin': [9, [10, 11]], 'archive_confirmed': True, 'segment_sealed': True, 'recovery_floor': [7, 8]}
check('valid operation', solve(r), True)
check('Deletion removes the next segment needed by a lagging replica', solve(dict(r, **{'slowest_replica': [10, [10, 12]]})), False)
check('A backup still needs the selected log segment', solve(dict(r, **{'backup_pin': [10, [10, 11]]})), False)
check('The only recoverable copy is deleted before archival', solve(dict(r, **{'archive_confirmed': False})), False)
check('Garbage collection deletes a segment still accepting writes', solve(dict(r, **{'segment_sealed': False})), False)
check('Deletion crosses the crash recovery replay floor', solve(dict(r, **{'recovery_floor': [8, 8]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'slowest_replica': [10, [10, 12]], 'backup_pin': [10, [10, 11]], 'archive_confirmed': False, 'segment_sealed': False, 'recovery_floor': [8, 8]}
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 wal retention operation is admitted even though garbage collection deletes a segment still accepting writes.

Return a Boolean admission decision for delete a write-ahead log segment. The record r must satisfy all of: r['slowest_replica'][0] < min(r['slowest_replica'][1]); r['backup_pin'][0] not in r['backup_pin'][1]; r['archive_confirmed'] is True; r['segment_sealed'] is True; r['recovery_floor'][0] < r['recovery_floor'][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['slowest_replica'][0] < min(r['slowest_replica'][1])) and (r['backup_pin'][0] not in r['backup_pin'][1]) and (r['archive_confirmed'] is True) and (r['segment_sealed'] is True)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'slowest_replica': [9, [10, 12]], 'backup_pin': [9, [10, 11]], 'archive_confirmed': True, 'segment_sealed': True, 'recovery_floor': [7, 8]}
check('valid operation', solve(r), True)
check('Deletion removes the next segment needed by a lagging replica', solve(dict(r, **{'slowest_replica': [10, [10, 12]]})), False)
check('A backup still needs the selected log segment', solve(dict(r, **{'backup_pin': [10, [10, 11]]})), False)
check('The only recoverable copy is deleted before archival', solve(dict(r, **{'archive_confirmed': False})), False)
check('Garbage collection deletes a segment still accepting writes', solve(dict(r, **{'segment_sealed': False})), False)
check('Deletion crosses the crash recovery replay floor', solve(dict(r, **{'recovery_floor': [8, 8]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'slowest_replica': [10, [10, 12]], 'backup_pin': [10, [10, 11]], 'archive_confirmed': False, 'segment_sealed': False, 'recovery_floor': [8, 8]}
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-wal-retention

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

N = 1
observations = []
def solve(r):
    return (r['backup_pin'][0] not in r['backup_pin'][1]) and (r['archive_confirmed'] is True) and (r['segment_sealed'] is True) and (r['recovery_floor'][0] < r['recovery_floor'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'slowest_replica': [9, [10, 12]], 'backup_pin': [9, [10, 11]], 'archive_confirmed': True, 'segment_sealed': True, 'recovery_floor': [7, 8]}
check('valid operation', solve(r), True)
check('Deletion removes the next segment needed by a lagging replica', solve(dict(r, **{'slowest_replica': [10, [10, 12]]})), False)
check('A backup still needs the selected log segment', solve(dict(r, **{'backup_pin': [10, [10, 11]]})), False)
check('The only recoverable copy is deleted before archival', solve(dict(r, **{'archive_confirmed': False})), False)
check('Garbage collection deletes a segment still accepting writes', solve(dict(r, **{'segment_sealed': False})), False)
check('Deletion crosses the crash recovery replay floor', solve(dict(r, **{'recovery_floor': [8, 8]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'slowest_replica': [10, [10, 12]], 'backup_pin': [10, [10, 11]], 'archive_confirmed': False, 'segment_sealed': False, 'recovery_floor': [8, 8]}
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 wal retention operation is admitted even though deletion crosses the crash recovery replay floor.

Return a Boolean admission decision for delete a write-ahead log segment. The record r must satisfy all of: r['slowest_replica'][0] < min(r['slowest_replica'][1]); r['backup_pin'][0] not in r['backup_pin'][1]; r['archive_confirmed'] is True; r['segment_sealed'] is True; r['recovery_floor'][0] < r['recovery_floor'][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 (sum(r['durable_acks'][0]) >= r['durable_acks'][1]) and (len(set(r['failure_domains'][0])) >= r['failure_domains'][1]) and (all(x == r['write_binding'][0] for x in r['write_binding'][1])) and (r['ack_deadline'][0] < r['ack_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'distinct_replicas': [['a', 'b'], 2], 'durable_acks': [[True, True, False], 2], 'failure_domains': [['z1', 'z2'], 2], 'write_binding': ['w2', ['w2', 'w2']], 'ack_deadline': [9, 10]}
check('valid operation', solve(r), True)
check('Repeated acknowledgements count as separate replicas', solve(dict(r, **{'distinct_replicas': [['a', 'a'], 2]})), False)
check('Volatile receipt acknowledgements satisfy durable concern', solve(dict(r, **{'durable_acks': [[True, False, False], 2]})), False)
check('Multiple copies in one failure domain satisfy zone concern', solve(dict(r, **{'failure_domains': [['z1', 'z1'], 2]})), False)
check('Replica acknowledgements refer to another write', solve(dict(r, **{'write_binding': ['w2', ['w1', 'w2']]})), False)
check('A write is reported successful after its response deadline', solve(dict(r, **{'ack_deadline': [10, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'distinct_replicas': [['a', 'a'], 2], 'durable_acks': [[True, False, False], 2], 'failure_domains': [['z1', 'z1'], 2], 'write_binding': ['w2', ['w1', 'w2']], 'ack_deadline': [10, 10]}
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-write-concern

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

N = 1
observations = []
def solve(r):
    return (len(set(r['distinct_replicas'][0])) >= r['distinct_replicas'][1]) and (len(set(r['failure_domains'][0])) >= r['failure_domains'][1]) and (all(x == r['write_binding'][0] for x in r['write_binding'][1])) and (r['ack_deadline'][0] < r['ack_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'distinct_replicas': [['a', 'b'], 2], 'durable_acks': [[True, True, False], 2], 'failure_domains': [['z1', 'z2'], 2], 'write_binding': ['w2', ['w2', 'w2']], 'ack_deadline': [9, 10]}
check('valid operation', solve(r), True)
check('Repeated acknowledgements count as separate replicas', solve(dict(r, **{'distinct_replicas': [['a', 'a'], 2]})), False)
check('Volatile receipt acknowledgements satisfy durable concern', solve(dict(r, **{'durable_acks': [[True, False, False], 2]})), False)
check('Multiple copies in one failure domain satisfy zone concern', solve(dict(r, **{'failure_domains': [['z1', 'z1'], 2]})), False)
check('Replica acknowledgements refer to another write', solve(dict(r, **{'write_binding': ['w2', ['w1', 'w2']]})), False)
check('A write is reported successful after its response deadline', solve(dict(r, **{'ack_deadline': [10, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'distinct_replicas': [['a', 'a'], 2], 'durable_acks': [[True, False, False], 2], 'failure_domains': [['z1', 'z1'], 2], 'write_binding': ['w2', ['w1', 'w2']], 'ack_deadline': [10, 10]}
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 write concern operation is admitted even though repeated acknowledgements count as separate replicas.

Return a Boolean admission decision for acknowledge a replicated write concern. The record r must satisfy all of: len(set(r['distinct_replicas'][0])) >= r['distinct_replicas'][1]; sum(r['durable_acks'][0]) >= r['durable_acks'][1]; len(set(r['failure_domains'][0])) >= r['failure_domains'][1]; all(x == r['write_binding'][0] for x in r['write_binding'][1]); r['ack_deadline'][0] < r['ack_deadline'][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 (len(set(r['distinct_replicas'][0])) >= r['distinct_replicas'][1]) and (len(set(r['failure_domains'][0])) >= r['failure_domains'][1]) and (all(x == r['write_binding'][0] for x in r['write_binding'][1])) and (r['ack_deadline'][0] < r['ack_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'distinct_replicas': [['a', 'b'], 2], 'durable_acks': [[True, True, False], 2], 'failure_domains': [['z1', 'z2'], 2], 'write_binding': ['w2', ['w2', 'w2']], 'ack_deadline': [9, 10]}
check('valid operation', solve(r), True)
check('Repeated acknowledgements count as separate replicas', solve(dict(r, **{'distinct_replicas': [['a', 'a'], 2]})), False)
check('Volatile receipt acknowledgements satisfy durable concern', solve(dict(r, **{'durable_acks': [[True, False, False], 2]})), False)
check('Multiple copies in one failure domain satisfy zone concern', solve(dict(r, **{'failure_domains': [['z1', 'z1'], 2]})), False)
check('Replica acknowledgements refer to another write', solve(dict(r, **{'write_binding': ['w2', ['w1', 'w2']]})), False)
check('A write is reported successful after its response deadline', solve(dict(r, **{'ack_deadline': [10, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'distinct_replicas': [['a', 'a'], 2], 'durable_acks': [[True, False, False], 2], 'failure_domains': [['z1', 'z1'], 2], 'write_binding': ['w2', ['w1', 'w2']], 'ack_deadline': [10, 10]}
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-write-concern

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

N = 1
observations = []
def solve(r):
    return (len(set(r['distinct_replicas'][0])) >= r['distinct_replicas'][1]) and (sum(r['durable_acks'][0]) >= r['durable_acks'][1]) and (all(x == r['write_binding'][0] for x in r['write_binding'][1])) and (r['ack_deadline'][0] < r['ack_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'distinct_replicas': [['a', 'b'], 2], 'durable_acks': [[True, True, False], 2], 'failure_domains': [['z1', 'z2'], 2], 'write_binding': ['w2', ['w2', 'w2']], 'ack_deadline': [9, 10]}
check('valid operation', solve(r), True)
check('Repeated acknowledgements count as separate replicas', solve(dict(r, **{'distinct_replicas': [['a', 'a'], 2]})), False)
check('Volatile receipt acknowledgements satisfy durable concern', solve(dict(r, **{'durable_acks': [[True, False, False], 2]})), False)
check('Multiple copies in one failure domain satisfy zone concern', solve(dict(r, **{'failure_domains': [['z1', 'z1'], 2]})), False)
check('Replica acknowledgements refer to another write', solve(dict(r, **{'write_binding': ['w2', ['w1', 'w2']]})), False)
check('A write is reported successful after its response deadline', solve(dict(r, **{'ack_deadline': [10, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'distinct_replicas': [['a', 'a'], 2], 'durable_acks': [[True, False, False], 2], 'failure_domains': [['z1', 'z1'], 2], 'write_binding': ['w2', ['w1', 'w2']], 'ack_deadline': [10, 10]}
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 write concern operation is admitted even though volatile receipt acknowledgements satisfy durable concern.

Return a Boolean admission decision for acknowledge a replicated write concern. The record r must satisfy all of: len(set(r['distinct_replicas'][0])) >= r['distinct_replicas'][1]; sum(r['durable_acks'][0]) >= r['durable_acks'][1]; len(set(r['failure_domains'][0])) >= r['failure_domains'][1]; all(x == r['write_binding'][0] for x in r['write_binding'][1]); r['ack_deadline'][0] < r['ack_deadline'][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 (len(set(r['distinct_replicas'][0])) >= r['distinct_replicas'][1]) and (sum(r['durable_acks'][0]) >= r['durable_acks'][1]) and (all(x == r['write_binding'][0] for x in r['write_binding'][1])) and (r['ack_deadline'][0] < r['ack_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'distinct_replicas': [['a', 'b'], 2], 'durable_acks': [[True, True, False], 2], 'failure_domains': [['z1', 'z2'], 2], 'write_binding': ['w2', ['w2', 'w2']], 'ack_deadline': [9, 10]}
check('valid operation', solve(r), True)
check('Repeated acknowledgements count as separate replicas', solve(dict(r, **{'distinct_replicas': [['a', 'a'], 2]})), False)
check('Volatile receipt acknowledgements satisfy durable concern', solve(dict(r, **{'durable_acks': [[True, False, False], 2]})), False)
check('Multiple copies in one failure domain satisfy zone concern', solve(dict(r, **{'failure_domains': [['z1', 'z1'], 2]})), False)
check('Replica acknowledgements refer to another write', solve(dict(r, **{'write_binding': ['w2', ['w1', 'w2']]})), False)
check('A write is reported successful after its response deadline', solve(dict(r, **{'ack_deadline': [10, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'distinct_replicas': [['a', 'a'], 2], 'durable_acks': [[True, False, False], 2], 'failure_domains': [['z1', 'z1'], 2], 'write_binding': ['w2', ['w1', 'w2']], 'ack_deadline': [10, 10]}
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-write-concern

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

N = 1
observations = []
def solve(r):
    return (len(set(r['distinct_replicas'][0])) >= r['distinct_replicas'][1]) and (sum(r['durable_acks'][0]) >= r['durable_acks'][1]) and (len(set(r['failure_domains'][0])) >= r['failure_domains'][1]) and (r['ack_deadline'][0] < r['ack_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'distinct_replicas': [['a', 'b'], 2], 'durable_acks': [[True, True, False], 2], 'failure_domains': [['z1', 'z2'], 2], 'write_binding': ['w2', ['w2', 'w2']], 'ack_deadline': [9, 10]}
check('valid operation', solve(r), True)
check('Repeated acknowledgements count as separate replicas', solve(dict(r, **{'distinct_replicas': [['a', 'a'], 2]})), False)
check('Volatile receipt acknowledgements satisfy durable concern', solve(dict(r, **{'durable_acks': [[True, False, False], 2]})), False)
check('Multiple copies in one failure domain satisfy zone concern', solve(dict(r, **{'failure_domains': [['z1', 'z1'], 2]})), False)
check('Replica acknowledgements refer to another write', solve(dict(r, **{'write_binding': ['w2', ['w1', 'w2']]})), False)
check('A write is reported successful after its response deadline', solve(dict(r, **{'ack_deadline': [10, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'distinct_replicas': [['a', 'a'], 2], 'durable_acks': [[True, False, False], 2], 'failure_domains': [['z1', 'z1'], 2], 'write_binding': ['w2', ['w1', 'w2']], 'ack_deadline': [10, 10]}
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 write concern operation is admitted even though multiple copies in one failure domain satisfy zone concern.

Return a Boolean admission decision for acknowledge a replicated write concern. The record r must satisfy all of: len(set(r['distinct_replicas'][0])) >= r['distinct_replicas'][1]; sum(r['durable_acks'][0]) >= r['durable_acks'][1]; len(set(r['failure_domains'][0])) >= r['failure_domains'][1]; all(x == r['write_binding'][0] for x in r['write_binding'][1]); r['ack_deadline'][0] < r['ack_deadline'][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 (len(set(r['distinct_replicas'][0])) >= r['distinct_replicas'][1]) and (sum(r['durable_acks'][0]) >= r['durable_acks'][1]) and (len(set(r['failure_domains'][0])) >= r['failure_domains'][1]) and (r['ack_deadline'][0] < r['ack_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'distinct_replicas': [['a', 'b'], 2], 'durable_acks': [[True, True, False], 2], 'failure_domains': [['z1', 'z2'], 2], 'write_binding': ['w2', ['w2', 'w2']], 'ack_deadline': [9, 10]}
check('valid operation', solve(r), True)
check('Repeated acknowledgements count as separate replicas', solve(dict(r, **{'distinct_replicas': [['a', 'a'], 2]})), False)
check('Volatile receipt acknowledgements satisfy durable concern', solve(dict(r, **{'durable_acks': [[True, False, False], 2]})), False)
check('Multiple copies in one failure domain satisfy zone concern', solve(dict(r, **{'failure_domains': [['z1', 'z1'], 2]})), False)
check('Replica acknowledgements refer to another write', solve(dict(r, **{'write_binding': ['w2', ['w1', 'w2']]})), False)
check('A write is reported successful after its response deadline', solve(dict(r, **{'ack_deadline': [10, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'distinct_replicas': [['a', 'a'], 2], 'durable_acks': [[True, False, False], 2], 'failure_domains': [['z1', 'z1'], 2], 'write_binding': ['w2', ['w1', 'w2']], 'ack_deadline': [10, 10]}
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-write-concern

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

N = 1
observations = []
def solve(r):
    return (len(set(r['distinct_replicas'][0])) >= r['distinct_replicas'][1]) and (sum(r['durable_acks'][0]) >= r['durable_acks'][1]) and (len(set(r['failure_domains'][0])) >= r['failure_domains'][1]) and (all(x == r['write_binding'][0] for x in r['write_binding'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'distinct_replicas': [['a', 'b'], 2], 'durable_acks': [[True, True, False], 2], 'failure_domains': [['z1', 'z2'], 2], 'write_binding': ['w2', ['w2', 'w2']], 'ack_deadline': [9, 10]}
check('valid operation', solve(r), True)
check('Repeated acknowledgements count as separate replicas', solve(dict(r, **{'distinct_replicas': [['a', 'a'], 2]})), False)
check('Volatile receipt acknowledgements satisfy durable concern', solve(dict(r, **{'durable_acks': [[True, False, False], 2]})), False)
check('Multiple copies in one failure domain satisfy zone concern', solve(dict(r, **{'failure_domains': [['z1', 'z1'], 2]})), False)
check('Replica acknowledgements refer to another write', solve(dict(r, **{'write_binding': ['w2', ['w1', 'w2']]})), False)
check('A write is reported successful after its response deadline', solve(dict(r, **{'ack_deadline': [10, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'distinct_replicas': [['a', 'a'], 2], 'durable_acks': [[True, False, False], 2], 'failure_domains': [['z1', 'z1'], 2], 'write_binding': ['w2', ['w1', 'w2']], 'ack_deadline': [10, 10]}
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 write concern operation is admitted even though replica acknowledgements refer to another write.

Return a Boolean admission decision for acknowledge a replicated write concern. The record r must satisfy all of: len(set(r['distinct_replicas'][0])) >= r['distinct_replicas'][1]; sum(r['durable_acks'][0]) >= r['durable_acks'][1]; len(set(r['failure_domains'][0])) >= r['failure_domains'][1]; all(x == r['write_binding'][0] for x in r['write_binding'][1]); r['ack_deadline'][0] < r['ack_deadline'][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 (len(set(r['distinct_replicas'][0])) >= r['distinct_replicas'][1]) and (sum(r['durable_acks'][0]) >= r['durable_acks'][1]) and (len(set(r['failure_domains'][0])) >= r['failure_domains'][1]) and (all(x == r['write_binding'][0] for x in r['write_binding'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'distinct_replicas': [['a', 'b'], 2], 'durable_acks': [[True, True, False], 2], 'failure_domains': [['z1', 'z2'], 2], 'write_binding': ['w2', ['w2', 'w2']], 'ack_deadline': [9, 10]}
check('valid operation', solve(r), True)
check('Repeated acknowledgements count as separate replicas', solve(dict(r, **{'distinct_replicas': [['a', 'a'], 2]})), False)
check('Volatile receipt acknowledgements satisfy durable concern', solve(dict(r, **{'durable_acks': [[True, False, False], 2]})), False)
check('Multiple copies in one failure domain satisfy zone concern', solve(dict(r, **{'failure_domains': [['z1', 'z1'], 2]})), False)
check('Replica acknowledgements refer to another write', solve(dict(r, **{'write_binding': ['w2', ['w1', 'w2']]})), False)
check('A write is reported successful after its response deadline', solve(dict(r, **{'ack_deadline': [10, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'distinct_replicas': [['a', 'a'], 2], 'durable_acks': [[True, False, False], 2], 'failure_domains': [['z1', 'z1'], 2], 'write_binding': ['w2', ['w1', 'w2']], 'ack_deadline': [10, 10]}
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-write-concern

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

N = 1
observations = []
def solve(r):
    return (sum(r['durable_acks'][0]) >= r['durable_acks'][1]) and (len(set(r['failure_domains'][0])) >= r['failure_domains'][1]) and (all(x == r['write_binding'][0] for x in r['write_binding'][1])) and (r['ack_deadline'][0] < r['ack_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'distinct_replicas': [['a', 'b'], 2], 'durable_acks': [[True, True, False], 2], 'failure_domains': [['z1', 'z2'], 2], 'write_binding': ['w2', ['w2', 'w2']], 'ack_deadline': [9, 10]}
check('valid operation', solve(r), True)
check('Repeated acknowledgements count as separate replicas', solve(dict(r, **{'distinct_replicas': [['a', 'a'], 2]})), False)
check('Volatile receipt acknowledgements satisfy durable concern', solve(dict(r, **{'durable_acks': [[True, False, False], 2]})), False)
check('Multiple copies in one failure domain satisfy zone concern', solve(dict(r, **{'failure_domains': [['z1', 'z1'], 2]})), False)
check('Replica acknowledgements refer to another write', solve(dict(r, **{'write_binding': ['w2', ['w1', 'w2']]})), False)
check('A write is reported successful after its response deadline', solve(dict(r, **{'ack_deadline': [10, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'distinct_replicas': [['a', 'a'], 2], 'durable_acks': [[True, False, False], 2], 'failure_domains': [['z1', 'z1'], 2], 'write_binding': ['w2', ['w1', 'w2']], 'ack_deadline': [10, 10]}
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 write concern operation is admitted even though a write is reported successful after its response deadline.

Return a Boolean admission decision for acknowledge a replicated write concern. The record r must satisfy all of: len(set(r['distinct_replicas'][0])) >= r['distinct_replicas'][1]; sum(r['durable_acks'][0]) >= r['durable_acks'][1]; len(set(r['failure_domains'][0])) >= r['failure_domains'][1]; all(x == r['write_binding'][0] for x in r['write_binding'][1]); r['ack_deadline'][0] < r['ack_deadline'][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['queue_capacity'][0]+r['queue_capacity'][1] <= r['queue_capacity'][2]) and (bool(r['routing_target'])) and (r['producer_epoch'][0] == r['producer_epoch'][1]) and (r['sequence_next'][0] == r['sequence_next'][1]+1)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'message_size': ['abc', 3], 'queue_capacity': [8, 2, 10], 'routing_target': ['orders'], 'producer_epoch': [4, 4], 'sequence_next': [8, 7]}
check('valid operation', solve(r), True)
check('An oversized message bypasses broker frame limits', solve(dict(r, **{'message_size': ['abcd', 3]})), False)
check('Admission exceeds the queue retained-byte budget', solve(dict(r, **{'queue_capacity': [8, 3, 10]})), False)
check('A mandatory message has no matching queue', solve(dict(r, **{'routing_target': []})), False)
check('A fenced producer continues publishing', solve(dict(r, **{'producer_epoch': [3, 4]})), False)
check('A producer sequence gap is accepted as contiguous', solve(dict(r, **{'sequence_next': [9, 7]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'message_size': ['abcd', 3], 'queue_capacity': [8, 3, 10], 'routing_target': [], 'producer_epoch': [3, 4], 'sequence_next': [9, 7]}
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-publish

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

N = 1
observations = []
def solve(r):
    return (len(r['message_size'][0]) <= r['message_size'][1]) and (bool(r['routing_target'])) and (r['producer_epoch'][0] == r['producer_epoch'][1]) and (r['sequence_next'][0] == r['sequence_next'][1]+1)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'message_size': ['abc', 3], 'queue_capacity': [8, 2, 10], 'routing_target': ['orders'], 'producer_epoch': [4, 4], 'sequence_next': [8, 7]}
check('valid operation', solve(r), True)
check('An oversized message bypasses broker frame limits', solve(dict(r, **{'message_size': ['abcd', 3]})), False)
check('Admission exceeds the queue retained-byte budget', solve(dict(r, **{'queue_capacity': [8, 3, 10]})), False)
check('A mandatory message has no matching queue', solve(dict(r, **{'routing_target': []})), False)
check('A fenced producer continues publishing', solve(dict(r, **{'producer_epoch': [3, 4]})), False)
check('A producer sequence gap is accepted as contiguous', solve(dict(r, **{'sequence_next': [9, 7]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'message_size': ['abcd', 3], 'queue_capacity': [8, 3, 10], 'routing_target': [], 'producer_epoch': [3, 4], 'sequence_next': [9, 7]}
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 publish operation is admitted even though an oversized message bypasses broker frame limits.

Return a Boolean admission decision for accept a durable queue publication. The record r must satisfy all of: len(r['message_size'][0]) <= r['message_size'][1]; r['queue_capacity'][0]+r['queue_capacity'][1] <= r['queue_capacity'][2]; bool(r['routing_target']); r['producer_epoch'][0] == r['producer_epoch'][1]; r['sequence_next'][0] == r['sequence_next'][1]+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 (len(r['message_size'][0]) <= r['message_size'][1]) and (bool(r['routing_target'])) and (r['producer_epoch'][0] == r['producer_epoch'][1]) and (r['sequence_next'][0] == r['sequence_next'][1]+1)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'message_size': ['abc', 3], 'queue_capacity': [8, 2, 10], 'routing_target': ['orders'], 'producer_epoch': [4, 4], 'sequence_next': [8, 7]}
check('valid operation', solve(r), True)
check('An oversized message bypasses broker frame limits', solve(dict(r, **{'message_size': ['abcd', 3]})), False)
check('Admission exceeds the queue retained-byte budget', solve(dict(r, **{'queue_capacity': [8, 3, 10]})), False)
check('A mandatory message has no matching queue', solve(dict(r, **{'routing_target': []})), False)
check('A fenced producer continues publishing', solve(dict(r, **{'producer_epoch': [3, 4]})), False)
check('A producer sequence gap is accepted as contiguous', solve(dict(r, **{'sequence_next': [9, 7]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'message_size': ['abcd', 3], 'queue_capacity': [8, 3, 10], 'routing_target': [], 'producer_epoch': [3, 4], 'sequence_next': [9, 7]}
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-publish

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

N = 1
observations = []
def solve(r):
    return (len(r['message_size'][0]) <= r['message_size'][1]) and (r['queue_capacity'][0]+r['queue_capacity'][1] <= r['queue_capacity'][2]) and (r['producer_epoch'][0] == r['producer_epoch'][1]) and (r['sequence_next'][0] == r['sequence_next'][1]+1)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'message_size': ['abc', 3], 'queue_capacity': [8, 2, 10], 'routing_target': ['orders'], 'producer_epoch': [4, 4], 'sequence_next': [8, 7]}
check('valid operation', solve(r), True)
check('An oversized message bypasses broker frame limits', solve(dict(r, **{'message_size': ['abcd', 3]})), False)
check('Admission exceeds the queue retained-byte budget', solve(dict(r, **{'queue_capacity': [8, 3, 10]})), False)
check('A mandatory message has no matching queue', solve(dict(r, **{'routing_target': []})), False)
check('A fenced producer continues publishing', solve(dict(r, **{'producer_epoch': [3, 4]})), False)
check('A producer sequence gap is accepted as contiguous', solve(dict(r, **{'sequence_next': [9, 7]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'message_size': ['abcd', 3], 'queue_capacity': [8, 3, 10], 'routing_target': [], 'producer_epoch': [3, 4], 'sequence_next': [9, 7]}
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 publish operation is admitted even though admission exceeds the queue retained-byte budget.

Return a Boolean admission decision for accept a durable queue publication. The record r must satisfy all of: len(r['message_size'][0]) <= r['message_size'][1]; r['queue_capacity'][0]+r['queue_capacity'][1] <= r['queue_capacity'][2]; bool(r['routing_target']); r['producer_epoch'][0] == r['producer_epoch'][1]; r['sequence_next'][0] == r['sequence_next'][1]+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 (len(r['message_size'][0]) <= r['message_size'][1]) and (r['queue_capacity'][0]+r['queue_capacity'][1] <= r['queue_capacity'][2]) and (r['producer_epoch'][0] == r['producer_epoch'][1]) and (r['sequence_next'][0] == r['sequence_next'][1]+1)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'message_size': ['abc', 3], 'queue_capacity': [8, 2, 10], 'routing_target': ['orders'], 'producer_epoch': [4, 4], 'sequence_next': [8, 7]}
check('valid operation', solve(r), True)
check('An oversized message bypasses broker frame limits', solve(dict(r, **{'message_size': ['abcd', 3]})), False)
check('Admission exceeds the queue retained-byte budget', solve(dict(r, **{'queue_capacity': [8, 3, 10]})), False)
check('A mandatory message has no matching queue', solve(dict(r, **{'routing_target': []})), False)
check('A fenced producer continues publishing', solve(dict(r, **{'producer_epoch': [3, 4]})), False)
check('A producer sequence gap is accepted as contiguous', solve(dict(r, **{'sequence_next': [9, 7]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'message_size': ['abcd', 3], 'queue_capacity': [8, 3, 10], 'routing_target': [], 'producer_epoch': [3, 4], 'sequence_next': [9, 7]}
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-publish

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

N = 1
observations = []
def solve(r):
    return (len(r['message_size'][0]) <= r['message_size'][1]) and (r['queue_capacity'][0]+r['queue_capacity'][1] <= r['queue_capacity'][2]) and (bool(r['routing_target'])) and (r['sequence_next'][0] == r['sequence_next'][1]+1)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'message_size': ['abc', 3], 'queue_capacity': [8, 2, 10], 'routing_target': ['orders'], 'producer_epoch': [4, 4], 'sequence_next': [8, 7]}
check('valid operation', solve(r), True)
check('An oversized message bypasses broker frame limits', solve(dict(r, **{'message_size': ['abcd', 3]})), False)
check('Admission exceeds the queue retained-byte budget', solve(dict(r, **{'queue_capacity': [8, 3, 10]})), False)
check('A mandatory message has no matching queue', solve(dict(r, **{'routing_target': []})), False)
check('A fenced producer continues publishing', solve(dict(r, **{'producer_epoch': [3, 4]})), False)
check('A producer sequence gap is accepted as contiguous', solve(dict(r, **{'sequence_next': [9, 7]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'message_size': ['abcd', 3], 'queue_capacity': [8, 3, 10], 'routing_target': [], 'producer_epoch': [3, 4], 'sequence_next': [9, 7]}
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 publish operation is admitted even though a mandatory message has no matching queue.

Return a Boolean admission decision for accept a durable queue publication. The record r must satisfy all of: len(r['message_size'][0]) <= r['message_size'][1]; r['queue_capacity'][0]+r['queue_capacity'][1] <= r['queue_capacity'][2]; bool(r['routing_target']); r['producer_epoch'][0] == r['producer_epoch'][1]; r['sequence_next'][0] == r['sequence_next'][1]+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 (len(r['message_size'][0]) <= r['message_size'][1]) and (r['queue_capacity'][0]+r['queue_capacity'][1] <= r['queue_capacity'][2]) and (bool(r['routing_target'])) and (r['sequence_next'][0] == r['sequence_next'][1]+1)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'message_size': ['abc', 3], 'queue_capacity': [8, 2, 10], 'routing_target': ['orders'], 'producer_epoch': [4, 4], 'sequence_next': [8, 7]}
check('valid operation', solve(r), True)
check('An oversized message bypasses broker frame limits', solve(dict(r, **{'message_size': ['abcd', 3]})), False)
check('Admission exceeds the queue retained-byte budget', solve(dict(r, **{'queue_capacity': [8, 3, 10]})), False)
check('A mandatory message has no matching queue', solve(dict(r, **{'routing_target': []})), False)
check('A fenced producer continues publishing', solve(dict(r, **{'producer_epoch': [3, 4]})), False)
check('A producer sequence gap is accepted as contiguous', solve(dict(r, **{'sequence_next': [9, 7]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'message_size': ['abcd', 3], 'queue_capacity': [8, 3, 10], 'routing_target': [], 'producer_epoch': [3, 4], 'sequence_next': [9, 7]}
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-publish

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

N = 1
observations = []
def solve(r):
    return (len(r['message_size'][0]) <= r['message_size'][1]) and (r['queue_capacity'][0]+r['queue_capacity'][1] <= r['queue_capacity'][2]) and (bool(r['routing_target'])) and (r['producer_epoch'][0] == r['producer_epoch'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'message_size': ['abc', 3], 'queue_capacity': [8, 2, 10], 'routing_target': ['orders'], 'producer_epoch': [4, 4], 'sequence_next': [8, 7]}
check('valid operation', solve(r), True)
check('An oversized message bypasses broker frame limits', solve(dict(r, **{'message_size': ['abcd', 3]})), False)
check('Admission exceeds the queue retained-byte budget', solve(dict(r, **{'queue_capacity': [8, 3, 10]})), False)
check('A mandatory message has no matching queue', solve(dict(r, **{'routing_target': []})), False)
check('A fenced producer continues publishing', solve(dict(r, **{'producer_epoch': [3, 4]})), False)
check('A producer sequence gap is accepted as contiguous', solve(dict(r, **{'sequence_next': [9, 7]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'message_size': ['abcd', 3], 'queue_capacity': [8, 3, 10], 'routing_target': [], 'producer_epoch': [3, 4], 'sequence_next': [9, 7]}
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 publish operation is admitted even though a fenced producer continues publishing.

Return a Boolean admission decision for accept a durable queue publication. The record r must satisfy all of: len(r['message_size'][0]) <= r['message_size'][1]; r['queue_capacity'][0]+r['queue_capacity'][1] <= r['queue_capacity'][2]; bool(r['routing_target']); r['producer_epoch'][0] == r['producer_epoch'][1]; r['sequence_next'][0] == r['sequence_next'][1]+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 (len(r['message_size'][0]) <= r['message_size'][1]) and (r['queue_capacity'][0]+r['queue_capacity'][1] <= r['queue_capacity'][2]) and (bool(r['routing_target'])) and (r['producer_epoch'][0] == r['producer_epoch'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'message_size': ['abc', 3], 'queue_capacity': [8, 2, 10], 'routing_target': ['orders'], 'producer_epoch': [4, 4], 'sequence_next': [8, 7]}
check('valid operation', solve(r), True)
check('An oversized message bypasses broker frame limits', solve(dict(r, **{'message_size': ['abcd', 3]})), False)
check('Admission exceeds the queue retained-byte budget', solve(dict(r, **{'queue_capacity': [8, 3, 10]})), False)
check('A mandatory message has no matching queue', solve(dict(r, **{'routing_target': []})), False)
check('A fenced producer continues publishing', solve(dict(r, **{'producer_epoch': [3, 4]})), False)
check('A producer sequence gap is accepted as contiguous', solve(dict(r, **{'sequence_next': [9, 7]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'message_size': ['abcd', 3], 'queue_capacity': [8, 3, 10], 'routing_target': [], 'producer_epoch': [3, 4], 'sequence_next': [9, 7]}
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-publish

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

N = 1
observations = []
def solve(r):
    return (r['queue_capacity'][0]+r['queue_capacity'][1] <= r['queue_capacity'][2]) and (bool(r['routing_target'])) and (r['producer_epoch'][0] == r['producer_epoch'][1]) and (r['sequence_next'][0] == r['sequence_next'][1]+1)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'message_size': ['abc', 3], 'queue_capacity': [8, 2, 10], 'routing_target': ['orders'], 'producer_epoch': [4, 4], 'sequence_next': [8, 7]}
check('valid operation', solve(r), True)
check('An oversized message bypasses broker frame limits', solve(dict(r, **{'message_size': ['abcd', 3]})), False)
check('Admission exceeds the queue retained-byte budget', solve(dict(r, **{'queue_capacity': [8, 3, 10]})), False)
check('A mandatory message has no matching queue', solve(dict(r, **{'routing_target': []})), False)
check('A fenced producer continues publishing', solve(dict(r, **{'producer_epoch': [3, 4]})), False)
check('A producer sequence gap is accepted as contiguous', solve(dict(r, **{'sequence_next': [9, 7]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'message_size': ['abcd', 3], 'queue_capacity': [8, 3, 10], 'routing_target': [], 'producer_epoch': [3, 4], 'sequence_next': [9, 7]}
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 publish operation is admitted even though a producer sequence gap is accepted as contiguous.

Return a Boolean admission decision for accept a durable queue publication. The record r must satisfy all of: len(r['message_size'][0]) <= r['message_size'][1]; r['queue_capacity'][0]+r['queue_capacity'][1] <= r['queue_capacity'][2]; bool(r['routing_target']); r['producer_epoch'][0] == r['producer_epoch'][1]; r['sequence_next'][0] == r['sequence_next'][1]+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['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)

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['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 scheduled message is delivered before its visibility instant.

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['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)

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['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 message is delivered after its expiration deadline.

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