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

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

evaluation_group: xd-partition-rebalance

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

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

prompt:
The partition rebalance operation is admitted even though a partition is assigned to a consumer not subscribed to its topic.

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

split: open-access


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

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

meaning: Fixture-count band, not measured model difficulty

deps: stdlib

evaluation_group: xd-partition-rebalance

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

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

prompt:
The partition rebalance operation is admitted even though rebalance exceeds the new owner partition capacity.

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

split: open-access


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

N = 1
observations = []
def solve(r):
    return (r['output_committed'] is True) and (r['offset_exact'][0] == r['offset_exact'][1]+1) and (r['transaction_open'] == 'open') and (r['producer_fenced'][0] == r['producer_fenced'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'input_owned': [8, 8], 'output_committed': True, 'offset_exact': [11, 10], 'transaction_open': 'open', 'producer_fenced': [9, 9]}
check('valid operation', solve(r), True)
check('Commit acknowledges an input partition no longer owned', solve(dict(r, **{'input_owned': [7, 8]})), False)
check('Input progress commits before the transformed output', solve(dict(r, **{'output_committed': False})), False)
check('Commit skips an unprocessed input offset', solve(dict(r, **{'offset_exact': [12, 10]})), False)
check('A commit revives an aborted consume transaction', solve(dict(r, **{'transaction_open': 'aborted'})), False)
check('A previous producer instance commits an old transaction', solve(dict(r, **{'producer_fenced': [8, 9]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'input_owned': [7, 8], 'output_committed': False, 'offset_exact': [12, 10], 'transaction_open': 'aborted', 'producer_fenced': [8, 9]}
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-transactional-consume

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

N = 1
observations = []
def solve(r):
    return (r['input_owned'][0] == r['input_owned'][1]) and (r['offset_exact'][0] == r['offset_exact'][1]+1) and (r['transaction_open'] == 'open') and (r['producer_fenced'][0] == r['producer_fenced'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'input_owned': [8, 8], 'output_committed': True, 'offset_exact': [11, 10], 'transaction_open': 'open', 'producer_fenced': [9, 9]}
check('valid operation', solve(r), True)
check('Commit acknowledges an input partition no longer owned', solve(dict(r, **{'input_owned': [7, 8]})), False)
check('Input progress commits before the transformed output', solve(dict(r, **{'output_committed': False})), False)
check('Commit skips an unprocessed input offset', solve(dict(r, **{'offset_exact': [12, 10]})), False)
check('A commit revives an aborted consume transaction', solve(dict(r, **{'transaction_open': 'aborted'})), False)
check('A previous producer instance commits an old transaction', solve(dict(r, **{'producer_fenced': [8, 9]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'input_owned': [7, 8], 'output_committed': False, 'offset_exact': [12, 10], 'transaction_open': 'aborted', 'producer_fenced': [8, 9]}
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 transactional consume operation is admitted even though commit acknowledges an input partition no longer owned.

Return a Boolean admission decision for commit a consume-transform-publish transaction. The record r must satisfy all of: r['input_owned'][0] == r['input_owned'][1]; r['output_committed'] is True; r['offset_exact'][0] == r['offset_exact'][1]+1; r['transaction_open'] == 'open'; r['producer_fenced'][0] == r['producer_fenced'][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['input_owned'][0] == r['input_owned'][1]) and (r['offset_exact'][0] == r['offset_exact'][1]+1) and (r['transaction_open'] == 'open') and (r['producer_fenced'][0] == r['producer_fenced'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'input_owned': [8, 8], 'output_committed': True, 'offset_exact': [11, 10], 'transaction_open': 'open', 'producer_fenced': [9, 9]}
check('valid operation', solve(r), True)
check('Commit acknowledges an input partition no longer owned', solve(dict(r, **{'input_owned': [7, 8]})), False)
check('Input progress commits before the transformed output', solve(dict(r, **{'output_committed': False})), False)
check('Commit skips an unprocessed input offset', solve(dict(r, **{'offset_exact': [12, 10]})), False)
check('A commit revives an aborted consume transaction', solve(dict(r, **{'transaction_open': 'aborted'})), False)
check('A previous producer instance commits an old transaction', solve(dict(r, **{'producer_fenced': [8, 9]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'input_owned': [7, 8], 'output_committed': False, 'offset_exact': [12, 10], 'transaction_open': 'aborted', 'producer_fenced': [8, 9]}
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-transactional-consume

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

N = 1
observations = []
def solve(r):
    return (r['input_owned'][0] == r['input_owned'][1]) and (r['output_committed'] is True) and (r['transaction_open'] == 'open') and (r['producer_fenced'][0] == r['producer_fenced'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'input_owned': [8, 8], 'output_committed': True, 'offset_exact': [11, 10], 'transaction_open': 'open', 'producer_fenced': [9, 9]}
check('valid operation', solve(r), True)
check('Commit acknowledges an input partition no longer owned', solve(dict(r, **{'input_owned': [7, 8]})), False)
check('Input progress commits before the transformed output', solve(dict(r, **{'output_committed': False})), False)
check('Commit skips an unprocessed input offset', solve(dict(r, **{'offset_exact': [12, 10]})), False)
check('A commit revives an aborted consume transaction', solve(dict(r, **{'transaction_open': 'aborted'})), False)
check('A previous producer instance commits an old transaction', solve(dict(r, **{'producer_fenced': [8, 9]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'input_owned': [7, 8], 'output_committed': False, 'offset_exact': [12, 10], 'transaction_open': 'aborted', 'producer_fenced': [8, 9]}
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 transactional consume operation is admitted even though input progress commits before the transformed output.

Return a Boolean admission decision for commit a consume-transform-publish transaction. The record r must satisfy all of: r['input_owned'][0] == r['input_owned'][1]; r['output_committed'] is True; r['offset_exact'][0] == r['offset_exact'][1]+1; r['transaction_open'] == 'open'; r['producer_fenced'][0] == r['producer_fenced'][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['input_owned'][0] == r['input_owned'][1]) and (r['output_committed'] is True) and (r['transaction_open'] == 'open') and (r['producer_fenced'][0] == r['producer_fenced'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'input_owned': [8, 8], 'output_committed': True, 'offset_exact': [11, 10], 'transaction_open': 'open', 'producer_fenced': [9, 9]}
check('valid operation', solve(r), True)
check('Commit acknowledges an input partition no longer owned', solve(dict(r, **{'input_owned': [7, 8]})), False)
check('Input progress commits before the transformed output', solve(dict(r, **{'output_committed': False})), False)
check('Commit skips an unprocessed input offset', solve(dict(r, **{'offset_exact': [12, 10]})), False)
check('A commit revives an aborted consume transaction', solve(dict(r, **{'transaction_open': 'aborted'})), False)
check('A previous producer instance commits an old transaction', solve(dict(r, **{'producer_fenced': [8, 9]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'input_owned': [7, 8], 'output_committed': False, 'offset_exact': [12, 10], 'transaction_open': 'aborted', 'producer_fenced': [8, 9]}
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-transactional-consume

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

N = 1
observations = []
def solve(r):
    return (r['input_owned'][0] == r['input_owned'][1]) and (r['output_committed'] is True) and (r['offset_exact'][0] == r['offset_exact'][1]+1) and (r['producer_fenced'][0] == r['producer_fenced'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'input_owned': [8, 8], 'output_committed': True, 'offset_exact': [11, 10], 'transaction_open': 'open', 'producer_fenced': [9, 9]}
check('valid operation', solve(r), True)
check('Commit acknowledges an input partition no longer owned', solve(dict(r, **{'input_owned': [7, 8]})), False)
check('Input progress commits before the transformed output', solve(dict(r, **{'output_committed': False})), False)
check('Commit skips an unprocessed input offset', solve(dict(r, **{'offset_exact': [12, 10]})), False)
check('A commit revives an aborted consume transaction', solve(dict(r, **{'transaction_open': 'aborted'})), False)
check('A previous producer instance commits an old transaction', solve(dict(r, **{'producer_fenced': [8, 9]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'input_owned': [7, 8], 'output_committed': False, 'offset_exact': [12, 10], 'transaction_open': 'aborted', 'producer_fenced': [8, 9]}
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 transactional consume operation is admitted even though commit skips an unprocessed input offset.

Return a Boolean admission decision for commit a consume-transform-publish transaction. The record r must satisfy all of: r['input_owned'][0] == r['input_owned'][1]; r['output_committed'] is True; r['offset_exact'][0] == r['offset_exact'][1]+1; r['transaction_open'] == 'open'; r['producer_fenced'][0] == r['producer_fenced'][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['input_owned'][0] == r['input_owned'][1]) and (r['output_committed'] is True) and (r['offset_exact'][0] == r['offset_exact'][1]+1) and (r['producer_fenced'][0] == r['producer_fenced'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'input_owned': [8, 8], 'output_committed': True, 'offset_exact': [11, 10], 'transaction_open': 'open', 'producer_fenced': [9, 9]}
check('valid operation', solve(r), True)
check('Commit acknowledges an input partition no longer owned', solve(dict(r, **{'input_owned': [7, 8]})), False)
check('Input progress commits before the transformed output', solve(dict(r, **{'output_committed': False})), False)
check('Commit skips an unprocessed input offset', solve(dict(r, **{'offset_exact': [12, 10]})), False)
check('A commit revives an aborted consume transaction', solve(dict(r, **{'transaction_open': 'aborted'})), False)
check('A previous producer instance commits an old transaction', solve(dict(r, **{'producer_fenced': [8, 9]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'input_owned': [7, 8], 'output_committed': False, 'offset_exact': [12, 10], 'transaction_open': 'aborted', 'producer_fenced': [8, 9]}
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-transactional-consume

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

N = 1
observations = []
def solve(r):
    return (r['input_owned'][0] == r['input_owned'][1]) and (r['output_committed'] is True) and (r['offset_exact'][0] == r['offset_exact'][1]+1) and (r['transaction_open'] == 'open')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'input_owned': [8, 8], 'output_committed': True, 'offset_exact': [11, 10], 'transaction_open': 'open', 'producer_fenced': [9, 9]}
check('valid operation', solve(r), True)
check('Commit acknowledges an input partition no longer owned', solve(dict(r, **{'input_owned': [7, 8]})), False)
check('Input progress commits before the transformed output', solve(dict(r, **{'output_committed': False})), False)
check('Commit skips an unprocessed input offset', solve(dict(r, **{'offset_exact': [12, 10]})), False)
check('A commit revives an aborted consume transaction', solve(dict(r, **{'transaction_open': 'aborted'})), False)
check('A previous producer instance commits an old transaction', solve(dict(r, **{'producer_fenced': [8, 9]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'input_owned': [7, 8], 'output_committed': False, 'offset_exact': [12, 10], 'transaction_open': 'aborted', 'producer_fenced': [8, 9]}
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 transactional consume operation is admitted even though a commit revives an aborted consume transaction.

Return a Boolean admission decision for commit a consume-transform-publish transaction. The record r must satisfy all of: r['input_owned'][0] == r['input_owned'][1]; r['output_committed'] is True; r['offset_exact'][0] == r['offset_exact'][1]+1; r['transaction_open'] == 'open'; r['producer_fenced'][0] == r['producer_fenced'][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['input_owned'][0] == r['input_owned'][1]) and (r['output_committed'] is True) and (r['offset_exact'][0] == r['offset_exact'][1]+1) and (r['transaction_open'] == 'open')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'input_owned': [8, 8], 'output_committed': True, 'offset_exact': [11, 10], 'transaction_open': 'open', 'producer_fenced': [9, 9]}
check('valid operation', solve(r), True)
check('Commit acknowledges an input partition no longer owned', solve(dict(r, **{'input_owned': [7, 8]})), False)
check('Input progress commits before the transformed output', solve(dict(r, **{'output_committed': False})), False)
check('Commit skips an unprocessed input offset', solve(dict(r, **{'offset_exact': [12, 10]})), False)
check('A commit revives an aborted consume transaction', solve(dict(r, **{'transaction_open': 'aborted'})), False)
check('A previous producer instance commits an old transaction', solve(dict(r, **{'producer_fenced': [8, 9]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'input_owned': [7, 8], 'output_committed': False, 'offset_exact': [12, 10], 'transaction_open': 'aborted', 'producer_fenced': [8, 9]}
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-transactional-consume

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

N = 1
observations = []
def solve(r):
    return (r['output_committed'] is True) and (r['offset_exact'][0] == r['offset_exact'][1]+1) and (r['transaction_open'] == 'open') and (r['producer_fenced'][0] == r['producer_fenced'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'input_owned': [8, 8], 'output_committed': True, 'offset_exact': [11, 10], 'transaction_open': 'open', 'producer_fenced': [9, 9]}
check('valid operation', solve(r), True)
check('Commit acknowledges an input partition no longer owned', solve(dict(r, **{'input_owned': [7, 8]})), False)
check('Input progress commits before the transformed output', solve(dict(r, **{'output_committed': False})), False)
check('Commit skips an unprocessed input offset', solve(dict(r, **{'offset_exact': [12, 10]})), False)
check('A commit revives an aborted consume transaction', solve(dict(r, **{'transaction_open': 'aborted'})), False)
check('A previous producer instance commits an old transaction', solve(dict(r, **{'producer_fenced': [8, 9]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'input_owned': [7, 8], 'output_committed': False, 'offset_exact': [12, 10], 'transaction_open': 'aborted', 'producer_fenced': [8, 9]}
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 transactional consume operation is admitted even though a previous producer instance commits an old transaction.

Return a Boolean admission decision for commit a consume-transform-publish transaction. The record r must satisfy all of: r['input_owned'][0] == r['input_owned'][1]; r['output_committed'] is True; r['offset_exact'][0] == r['offset_exact'][1]+1; r['transaction_open'] == 'open'; r['producer_fenced'][0] == r['producer_fenced'][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['reserved_protected'] is False) and (r['namespace_match'][0] == r['namespace_match'][1]) and (r['queue_incarnation'][0] == r['queue_incarnation'][1]) and (r['retention_hold'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'boundary_fixed': [100, 100], 'reserved_protected': False, 'namespace_match': ['t1', 't1'], 'queue_incarnation': ['g2', 'g2'], 'retention_hold': False}
check('valid operation', solve(r), True)
check('A moving purge boundary consumes messages published after purge began', solve(dict(r, **{'boundary_fixed': [101, 100]})), False)
check('Purge deletes an actively reserved delivery', solve(dict(r, **{'reserved_protected': True})), False)
check('Purge crosses into another tenant queue', solve(dict(r, **{'namespace_match': ['t2', 't1']})), False)
check('A delayed purge targets a recreated queue with the same name', solve(dict(r, **{'queue_incarnation': ['g1', 'g2']})), False)
check('Purge ignores an explicit retention hold', solve(dict(r, **{'retention_hold': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'boundary_fixed': [101, 100], 'reserved_protected': True, 'namespace_match': ['t2', 't1'], 'queue_incarnation': ['g1', 'g2'], 'retention_hold': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-purge

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

N = 1
observations = []
def solve(r):
    return (r['boundary_fixed'][0] <= r['boundary_fixed'][1]) and (r['namespace_match'][0] == r['namespace_match'][1]) and (r['queue_incarnation'][0] == r['queue_incarnation'][1]) and (r['retention_hold'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'boundary_fixed': [100, 100], 'reserved_protected': False, 'namespace_match': ['t1', 't1'], 'queue_incarnation': ['g2', 'g2'], 'retention_hold': False}
check('valid operation', solve(r), True)
check('A moving purge boundary consumes messages published after purge began', solve(dict(r, **{'boundary_fixed': [101, 100]})), False)
check('Purge deletes an actively reserved delivery', solve(dict(r, **{'reserved_protected': True})), False)
check('Purge crosses into another tenant queue', solve(dict(r, **{'namespace_match': ['t2', 't1']})), False)
check('A delayed purge targets a recreated queue with the same name', solve(dict(r, **{'queue_incarnation': ['g1', 'g2']})), False)
check('Purge ignores an explicit retention hold', solve(dict(r, **{'retention_hold': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'boundary_fixed': [101, 100], 'reserved_protected': True, 'namespace_match': ['t2', 't1'], 'queue_incarnation': ['g1', 'g2'], 'retention_hold': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue purge operation is admitted even though a moving purge boundary consumes messages published after purge began.

Return a Boolean admission decision for purge messages through a selected boundary. The record r must satisfy all of: r['boundary_fixed'][0] <= r['boundary_fixed'][1]; r['reserved_protected'] is False; r['namespace_match'][0] == r['namespace_match'][1]; r['queue_incarnation'][0] == r['queue_incarnation'][1]; r['retention_hold'] is False. 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['boundary_fixed'][0] <= r['boundary_fixed'][1]) and (r['namespace_match'][0] == r['namespace_match'][1]) and (r['queue_incarnation'][0] == r['queue_incarnation'][1]) and (r['retention_hold'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'boundary_fixed': [100, 100], 'reserved_protected': False, 'namespace_match': ['t1', 't1'], 'queue_incarnation': ['g2', 'g2'], 'retention_hold': False}
check('valid operation', solve(r), True)
check('A moving purge boundary consumes messages published after purge began', solve(dict(r, **{'boundary_fixed': [101, 100]})), False)
check('Purge deletes an actively reserved delivery', solve(dict(r, **{'reserved_protected': True})), False)
check('Purge crosses into another tenant queue', solve(dict(r, **{'namespace_match': ['t2', 't1']})), False)
check('A delayed purge targets a recreated queue with the same name', solve(dict(r, **{'queue_incarnation': ['g1', 'g2']})), False)
check('Purge ignores an explicit retention hold', solve(dict(r, **{'retention_hold': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'boundary_fixed': [101, 100], 'reserved_protected': True, 'namespace_match': ['t2', 't1'], 'queue_incarnation': ['g1', 'g2'], 'retention_hold': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-purge

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

N = 1
observations = []
def solve(r):
    return (r['boundary_fixed'][0] <= r['boundary_fixed'][1]) and (r['reserved_protected'] is False) and (r['queue_incarnation'][0] == r['queue_incarnation'][1]) and (r['retention_hold'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'boundary_fixed': [100, 100], 'reserved_protected': False, 'namespace_match': ['t1', 't1'], 'queue_incarnation': ['g2', 'g2'], 'retention_hold': False}
check('valid operation', solve(r), True)
check('A moving purge boundary consumes messages published after purge began', solve(dict(r, **{'boundary_fixed': [101, 100]})), False)
check('Purge deletes an actively reserved delivery', solve(dict(r, **{'reserved_protected': True})), False)
check('Purge crosses into another tenant queue', solve(dict(r, **{'namespace_match': ['t2', 't1']})), False)
check('A delayed purge targets a recreated queue with the same name', solve(dict(r, **{'queue_incarnation': ['g1', 'g2']})), False)
check('Purge ignores an explicit retention hold', solve(dict(r, **{'retention_hold': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'boundary_fixed': [101, 100], 'reserved_protected': True, 'namespace_match': ['t2', 't1'], 'queue_incarnation': ['g1', 'g2'], 'retention_hold': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue purge operation is admitted even though purge deletes an actively reserved delivery.

Return a Boolean admission decision for purge messages through a selected boundary. The record r must satisfy all of: r['boundary_fixed'][0] <= r['boundary_fixed'][1]; r['reserved_protected'] is False; r['namespace_match'][0] == r['namespace_match'][1]; r['queue_incarnation'][0] == r['queue_incarnation'][1]; r['retention_hold'] is False. 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['boundary_fixed'][0] <= r['boundary_fixed'][1]) and (r['reserved_protected'] is False) and (r['queue_incarnation'][0] == r['queue_incarnation'][1]) and (r['retention_hold'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'boundary_fixed': [100, 100], 'reserved_protected': False, 'namespace_match': ['t1', 't1'], 'queue_incarnation': ['g2', 'g2'], 'retention_hold': False}
check('valid operation', solve(r), True)
check('A moving purge boundary consumes messages published after purge began', solve(dict(r, **{'boundary_fixed': [101, 100]})), False)
check('Purge deletes an actively reserved delivery', solve(dict(r, **{'reserved_protected': True})), False)
check('Purge crosses into another tenant queue', solve(dict(r, **{'namespace_match': ['t2', 't1']})), False)
check('A delayed purge targets a recreated queue with the same name', solve(dict(r, **{'queue_incarnation': ['g1', 'g2']})), False)
check('Purge ignores an explicit retention hold', solve(dict(r, **{'retention_hold': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'boundary_fixed': [101, 100], 'reserved_protected': True, 'namespace_match': ['t2', 't1'], 'queue_incarnation': ['g1', 'g2'], 'retention_hold': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-purge

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

N = 1
observations = []
def solve(r):
    return (r['boundary_fixed'][0] <= r['boundary_fixed'][1]) and (r['reserved_protected'] is False) and (r['namespace_match'][0] == r['namespace_match'][1]) and (r['retention_hold'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'boundary_fixed': [100, 100], 'reserved_protected': False, 'namespace_match': ['t1', 't1'], 'queue_incarnation': ['g2', 'g2'], 'retention_hold': False}
check('valid operation', solve(r), True)
check('A moving purge boundary consumes messages published after purge began', solve(dict(r, **{'boundary_fixed': [101, 100]})), False)
check('Purge deletes an actively reserved delivery', solve(dict(r, **{'reserved_protected': True})), False)
check('Purge crosses into another tenant queue', solve(dict(r, **{'namespace_match': ['t2', 't1']})), False)
check('A delayed purge targets a recreated queue with the same name', solve(dict(r, **{'queue_incarnation': ['g1', 'g2']})), False)
check('Purge ignores an explicit retention hold', solve(dict(r, **{'retention_hold': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'boundary_fixed': [101, 100], 'reserved_protected': True, 'namespace_match': ['t2', 't1'], 'queue_incarnation': ['g1', 'g2'], 'retention_hold': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue purge operation is admitted even though purge crosses into another tenant queue.

Return a Boolean admission decision for purge messages through a selected boundary. The record r must satisfy all of: r['boundary_fixed'][0] <= r['boundary_fixed'][1]; r['reserved_protected'] is False; r['namespace_match'][0] == r['namespace_match'][1]; r['queue_incarnation'][0] == r['queue_incarnation'][1]; r['retention_hold'] is False. 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['boundary_fixed'][0] <= r['boundary_fixed'][1]) and (r['reserved_protected'] is False) and (r['namespace_match'][0] == r['namespace_match'][1]) and (r['retention_hold'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'boundary_fixed': [100, 100], 'reserved_protected': False, 'namespace_match': ['t1', 't1'], 'queue_incarnation': ['g2', 'g2'], 'retention_hold': False}
check('valid operation', solve(r), True)
check('A moving purge boundary consumes messages published after purge began', solve(dict(r, **{'boundary_fixed': [101, 100]})), False)
check('Purge deletes an actively reserved delivery', solve(dict(r, **{'reserved_protected': True})), False)
check('Purge crosses into another tenant queue', solve(dict(r, **{'namespace_match': ['t2', 't1']})), False)
check('A delayed purge targets a recreated queue with the same name', solve(dict(r, **{'queue_incarnation': ['g1', 'g2']})), False)
check('Purge ignores an explicit retention hold', solve(dict(r, **{'retention_hold': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'boundary_fixed': [101, 100], 'reserved_protected': True, 'namespace_match': ['t2', 't1'], 'queue_incarnation': ['g1', 'g2'], 'retention_hold': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-purge

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

N = 1
observations = []
def solve(r):
    return (r['boundary_fixed'][0] <= r['boundary_fixed'][1]) and (r['reserved_protected'] is False) and (r['namespace_match'][0] == r['namespace_match'][1]) and (r['queue_incarnation'][0] == r['queue_incarnation'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'boundary_fixed': [100, 100], 'reserved_protected': False, 'namespace_match': ['t1', 't1'], 'queue_incarnation': ['g2', 'g2'], 'retention_hold': False}
check('valid operation', solve(r), True)
check('A moving purge boundary consumes messages published after purge began', solve(dict(r, **{'boundary_fixed': [101, 100]})), False)
check('Purge deletes an actively reserved delivery', solve(dict(r, **{'reserved_protected': True})), False)
check('Purge crosses into another tenant queue', solve(dict(r, **{'namespace_match': ['t2', 't1']})), False)
check('A delayed purge targets a recreated queue with the same name', solve(dict(r, **{'queue_incarnation': ['g1', 'g2']})), False)
check('Purge ignores an explicit retention hold', solve(dict(r, **{'retention_hold': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'boundary_fixed': [101, 100], 'reserved_protected': True, 'namespace_match': ['t2', 't1'], 'queue_incarnation': ['g1', 'g2'], 'retention_hold': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue purge operation is admitted even though a delayed purge targets a recreated queue with the same name.

Return a Boolean admission decision for purge messages through a selected boundary. The record r must satisfy all of: r['boundary_fixed'][0] <= r['boundary_fixed'][1]; r['reserved_protected'] is False; r['namespace_match'][0] == r['namespace_match'][1]; r['queue_incarnation'][0] == r['queue_incarnation'][1]; r['retention_hold'] is False. 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['boundary_fixed'][0] <= r['boundary_fixed'][1]) and (r['reserved_protected'] is False) and (r['namespace_match'][0] == r['namespace_match'][1]) and (r['queue_incarnation'][0] == r['queue_incarnation'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'boundary_fixed': [100, 100], 'reserved_protected': False, 'namespace_match': ['t1', 't1'], 'queue_incarnation': ['g2', 'g2'], 'retention_hold': False}
check('valid operation', solve(r), True)
check('A moving purge boundary consumes messages published after purge began', solve(dict(r, **{'boundary_fixed': [101, 100]})), False)
check('Purge deletes an actively reserved delivery', solve(dict(r, **{'reserved_protected': True})), False)
check('Purge crosses into another tenant queue', solve(dict(r, **{'namespace_match': ['t2', 't1']})), False)
check('A delayed purge targets a recreated queue with the same name', solve(dict(r, **{'queue_incarnation': ['g1', 'g2']})), False)
check('Purge ignores an explicit retention hold', solve(dict(r, **{'retention_hold': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'boundary_fixed': [101, 100], 'reserved_protected': True, 'namespace_match': ['t2', 't1'], 'queue_incarnation': ['g1', 'g2'], 'retention_hold': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-queue-purge

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

N = 1
observations = []
def solve(r):
    return (r['reserved_protected'] is False) and (r['namespace_match'][0] == r['namespace_match'][1]) and (r['queue_incarnation'][0] == r['queue_incarnation'][1]) and (r['retention_hold'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'boundary_fixed': [100, 100], 'reserved_protected': False, 'namespace_match': ['t1', 't1'], 'queue_incarnation': ['g2', 'g2'], 'retention_hold': False}
check('valid operation', solve(r), True)
check('A moving purge boundary consumes messages published after purge began', solve(dict(r, **{'boundary_fixed': [101, 100]})), False)
check('Purge deletes an actively reserved delivery', solve(dict(r, **{'reserved_protected': True})), False)
check('Purge crosses into another tenant queue', solve(dict(r, **{'namespace_match': ['t2', 't1']})), False)
check('A delayed purge targets a recreated queue with the same name', solve(dict(r, **{'queue_incarnation': ['g1', 'g2']})), False)
check('Purge ignores an explicit retention hold', solve(dict(r, **{'retention_hold': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'boundary_fixed': [101, 100], 'reserved_protected': True, 'namespace_match': ['t2', 't1'], 'queue_incarnation': ['g1', 'g2'], 'retention_hold': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The queue purge operation is admitted even though purge ignores an explicit retention hold.

Return a Boolean admission decision for purge messages through a selected boundary. The record r must satisfy all of: r['boundary_fixed'][0] <= r['boundary_fixed'][1]; r['reserved_protected'] is False; r['namespace_match'][0] == r['namespace_match'][1]; r['queue_incarnation'][0] == r['queue_incarnation'][1]; r['retention_hold'] is False. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (r['attempt_budget'][0] < r['attempt_budget'][1]) and (r['remaining_deadline'][0]+r['remaining_deadline'][1] <= r['remaining_deadline'][2]) and (r['idempotency_safe'][0] or bool(r['idempotency_safe'][1])) and (r['retry_after'][0] >= r['retry_after'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'retryable_result': 503, 'attempt_budget': [2, 3], 'remaining_deadline': [5, 3, 8], 'idempotency_safe': [False, 'op1'], 'retry_after': [10, 10]}
check('valid operation', solve(r), True)
check('A permanent rejection triggers another attempt', solve(dict(r, **{'retryable_result': 400})), False)
check('An exhausted attempt budget permits one more request', solve(dict(r, **{'attempt_budget': [3, 3]})), False)
check('A retry cannot finish within its remaining request deadline', solve(dict(r, **{'remaining_deadline': [5, 4, 8]})), False)
check('A non-idempotent write is retried without an identity key', solve(dict(r, **{'idempotency_safe': [False, '']})), False)
check('A retry ignores the server requested wait period', solve(dict(r, **{'retry_after': [9, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'retryable_result': 400, 'attempt_budget': [3, 3], 'remaining_deadline': [5, 4, 8], 'idempotency_safe': [False, ''], 'retry_after': [9, 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-retry-admission

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

N = 1
observations = []
def solve(r):
    return (r['retryable_result'] in [408,429,500,502,503,504]) and (r['remaining_deadline'][0]+r['remaining_deadline'][1] <= r['remaining_deadline'][2]) and (r['idempotency_safe'][0] or bool(r['idempotency_safe'][1])) and (r['retry_after'][0] >= r['retry_after'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'retryable_result': 503, 'attempt_budget': [2, 3], 'remaining_deadline': [5, 3, 8], 'idempotency_safe': [False, 'op1'], 'retry_after': [10, 10]}
check('valid operation', solve(r), True)
check('A permanent rejection triggers another attempt', solve(dict(r, **{'retryable_result': 400})), False)
check('An exhausted attempt budget permits one more request', solve(dict(r, **{'attempt_budget': [3, 3]})), False)
check('A retry cannot finish within its remaining request deadline', solve(dict(r, **{'remaining_deadline': [5, 4, 8]})), False)
check('A non-idempotent write is retried without an identity key', solve(dict(r, **{'idempotency_safe': [False, '']})), False)
check('A retry ignores the server requested wait period', solve(dict(r, **{'retry_after': [9, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'retryable_result': 400, 'attempt_budget': [3, 3], 'remaining_deadline': [5, 4, 8], 'idempotency_safe': [False, ''], 'retry_after': [9, 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 retry admission operation is admitted even though a permanent rejection triggers another attempt.

Return a Boolean admission decision for admit a remote operation retry. The record r must satisfy all of: r['retryable_result'] in [408,429,500,502,503,504]; r['attempt_budget'][0] < r['attempt_budget'][1]; r['remaining_deadline'][0]+r['remaining_deadline'][1] <= r['remaining_deadline'][2]; r['idempotency_safe'][0] or bool(r['idempotency_safe'][1]); r['retry_after'][0] >= r['retry_after'][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['retryable_result'] in [408,429,500,502,503,504]) and (r['remaining_deadline'][0]+r['remaining_deadline'][1] <= r['remaining_deadline'][2]) and (r['idempotency_safe'][0] or bool(r['idempotency_safe'][1])) and (r['retry_after'][0] >= r['retry_after'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'retryable_result': 503, 'attempt_budget': [2, 3], 'remaining_deadline': [5, 3, 8], 'idempotency_safe': [False, 'op1'], 'retry_after': [10, 10]}
check('valid operation', solve(r), True)
check('A permanent rejection triggers another attempt', solve(dict(r, **{'retryable_result': 400})), False)
check('An exhausted attempt budget permits one more request', solve(dict(r, **{'attempt_budget': [3, 3]})), False)
check('A retry cannot finish within its remaining request deadline', solve(dict(r, **{'remaining_deadline': [5, 4, 8]})), False)
check('A non-idempotent write is retried without an identity key', solve(dict(r, **{'idempotency_safe': [False, '']})), False)
check('A retry ignores the server requested wait period', solve(dict(r, **{'retry_after': [9, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'retryable_result': 400, 'attempt_budget': [3, 3], 'remaining_deadline': [5, 4, 8], 'idempotency_safe': [False, ''], 'retry_after': [9, 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-retry-admission

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

N = 1
observations = []
def solve(r):
    return (r['retryable_result'] in [408,429,500,502,503,504]) and (r['attempt_budget'][0] < r['attempt_budget'][1]) and (r['idempotency_safe'][0] or bool(r['idempotency_safe'][1])) and (r['retry_after'][0] >= r['retry_after'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'retryable_result': 503, 'attempt_budget': [2, 3], 'remaining_deadline': [5, 3, 8], 'idempotency_safe': [False, 'op1'], 'retry_after': [10, 10]}
check('valid operation', solve(r), True)
check('A permanent rejection triggers another attempt', solve(dict(r, **{'retryable_result': 400})), False)
check('An exhausted attempt budget permits one more request', solve(dict(r, **{'attempt_budget': [3, 3]})), False)
check('A retry cannot finish within its remaining request deadline', solve(dict(r, **{'remaining_deadline': [5, 4, 8]})), False)
check('A non-idempotent write is retried without an identity key', solve(dict(r, **{'idempotency_safe': [False, '']})), False)
check('A retry ignores the server requested wait period', solve(dict(r, **{'retry_after': [9, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'retryable_result': 400, 'attempt_budget': [3, 3], 'remaining_deadline': [5, 4, 8], 'idempotency_safe': [False, ''], 'retry_after': [9, 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 retry admission operation is admitted even though an exhausted attempt budget permits one more request.

Return a Boolean admission decision for admit a remote operation retry. The record r must satisfy all of: r['retryable_result'] in [408,429,500,502,503,504]; r['attempt_budget'][0] < r['attempt_budget'][1]; r['remaining_deadline'][0]+r['remaining_deadline'][1] <= r['remaining_deadline'][2]; r['idempotency_safe'][0] or bool(r['idempotency_safe'][1]); r['retry_after'][0] >= r['retry_after'][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['retryable_result'] in [408,429,500,502,503,504]) and (r['attempt_budget'][0] < r['attempt_budget'][1]) and (r['idempotency_safe'][0] or bool(r['idempotency_safe'][1])) and (r['retry_after'][0] >= r['retry_after'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'retryable_result': 503, 'attempt_budget': [2, 3], 'remaining_deadline': [5, 3, 8], 'idempotency_safe': [False, 'op1'], 'retry_after': [10, 10]}
check('valid operation', solve(r), True)
check('A permanent rejection triggers another attempt', solve(dict(r, **{'retryable_result': 400})), False)
check('An exhausted attempt budget permits one more request', solve(dict(r, **{'attempt_budget': [3, 3]})), False)
check('A retry cannot finish within its remaining request deadline', solve(dict(r, **{'remaining_deadline': [5, 4, 8]})), False)
check('A non-idempotent write is retried without an identity key', solve(dict(r, **{'idempotency_safe': [False, '']})), False)
check('A retry ignores the server requested wait period', solve(dict(r, **{'retry_after': [9, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'retryable_result': 400, 'attempt_budget': [3, 3], 'remaining_deadline': [5, 4, 8], 'idempotency_safe': [False, ''], 'retry_after': [9, 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-retry-admission

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

N = 1
observations = []
def solve(r):
    return (r['retryable_result'] in [408,429,500,502,503,504]) and (r['attempt_budget'][0] < r['attempt_budget'][1]) and (r['remaining_deadline'][0]+r['remaining_deadline'][1] <= r['remaining_deadline'][2]) and (r['retry_after'][0] >= r['retry_after'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'retryable_result': 503, 'attempt_budget': [2, 3], 'remaining_deadline': [5, 3, 8], 'idempotency_safe': [False, 'op1'], 'retry_after': [10, 10]}
check('valid operation', solve(r), True)
check('A permanent rejection triggers another attempt', solve(dict(r, **{'retryable_result': 400})), False)
check('An exhausted attempt budget permits one more request', solve(dict(r, **{'attempt_budget': [3, 3]})), False)
check('A retry cannot finish within its remaining request deadline', solve(dict(r, **{'remaining_deadline': [5, 4, 8]})), False)
check('A non-idempotent write is retried without an identity key', solve(dict(r, **{'idempotency_safe': [False, '']})), False)
check('A retry ignores the server requested wait period', solve(dict(r, **{'retry_after': [9, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'retryable_result': 400, 'attempt_budget': [3, 3], 'remaining_deadline': [5, 4, 8], 'idempotency_safe': [False, ''], 'retry_after': [9, 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 retry admission operation is admitted even though a retry cannot finish within its remaining request deadline.

Return a Boolean admission decision for admit a remote operation retry. The record r must satisfy all of: r['retryable_result'] in [408,429,500,502,503,504]; r['attempt_budget'][0] < r['attempt_budget'][1]; r['remaining_deadline'][0]+r['remaining_deadline'][1] <= r['remaining_deadline'][2]; r['idempotency_safe'][0] or bool(r['idempotency_safe'][1]); r['retry_after'][0] >= r['retry_after'][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['retryable_result'] in [408,429,500,502,503,504]) and (r['attempt_budget'][0] < r['attempt_budget'][1]) and (r['remaining_deadline'][0]+r['remaining_deadline'][1] <= r['remaining_deadline'][2]) and (r['retry_after'][0] >= r['retry_after'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'retryable_result': 503, 'attempt_budget': [2, 3], 'remaining_deadline': [5, 3, 8], 'idempotency_safe': [False, 'op1'], 'retry_after': [10, 10]}
check('valid operation', solve(r), True)
check('A permanent rejection triggers another attempt', solve(dict(r, **{'retryable_result': 400})), False)
check('An exhausted attempt budget permits one more request', solve(dict(r, **{'attempt_budget': [3, 3]})), False)
check('A retry cannot finish within its remaining request deadline', solve(dict(r, **{'remaining_deadline': [5, 4, 8]})), False)
check('A non-idempotent write is retried without an identity key', solve(dict(r, **{'idempotency_safe': [False, '']})), False)
check('A retry ignores the server requested wait period', solve(dict(r, **{'retry_after': [9, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'retryable_result': 400, 'attempt_budget': [3, 3], 'remaining_deadline': [5, 4, 8], 'idempotency_safe': [False, ''], 'retry_after': [9, 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-retry-admission

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

N = 1
observations = []
def solve(r):
    return (r['retryable_result'] in [408,429,500,502,503,504]) and (r['attempt_budget'][0] < r['attempt_budget'][1]) and (r['remaining_deadline'][0]+r['remaining_deadline'][1] <= r['remaining_deadline'][2]) and (r['idempotency_safe'][0] or bool(r['idempotency_safe'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'retryable_result': 503, 'attempt_budget': [2, 3], 'remaining_deadline': [5, 3, 8], 'idempotency_safe': [False, 'op1'], 'retry_after': [10, 10]}
check('valid operation', solve(r), True)
check('A permanent rejection triggers another attempt', solve(dict(r, **{'retryable_result': 400})), False)
check('An exhausted attempt budget permits one more request', solve(dict(r, **{'attempt_budget': [3, 3]})), False)
check('A retry cannot finish within its remaining request deadline', solve(dict(r, **{'remaining_deadline': [5, 4, 8]})), False)
check('A non-idempotent write is retried without an identity key', solve(dict(r, **{'idempotency_safe': [False, '']})), False)
check('A retry ignores the server requested wait period', solve(dict(r, **{'retry_after': [9, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'retryable_result': 400, 'attempt_budget': [3, 3], 'remaining_deadline': [5, 4, 8], 'idempotency_safe': [False, ''], 'retry_after': [9, 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 retry admission operation is admitted even though a non-idempotent write is retried without an identity key.

Return a Boolean admission decision for admit a remote operation retry. The record r must satisfy all of: r['retryable_result'] in [408,429,500,502,503,504]; r['attempt_budget'][0] < r['attempt_budget'][1]; r['remaining_deadline'][0]+r['remaining_deadline'][1] <= r['remaining_deadline'][2]; r['idempotency_safe'][0] or bool(r['idempotency_safe'][1]); r['retry_after'][0] >= r['retry_after'][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['retryable_result'] in [408,429,500,502,503,504]) and (r['attempt_budget'][0] < r['attempt_budget'][1]) and (r['remaining_deadline'][0]+r['remaining_deadline'][1] <= r['remaining_deadline'][2]) and (r['idempotency_safe'][0] or bool(r['idempotency_safe'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'retryable_result': 503, 'attempt_budget': [2, 3], 'remaining_deadline': [5, 3, 8], 'idempotency_safe': [False, 'op1'], 'retry_after': [10, 10]}
check('valid operation', solve(r), True)
check('A permanent rejection triggers another attempt', solve(dict(r, **{'retryable_result': 400})), False)
check('An exhausted attempt budget permits one more request', solve(dict(r, **{'attempt_budget': [3, 3]})), False)
check('A retry cannot finish within its remaining request deadline', solve(dict(r, **{'remaining_deadline': [5, 4, 8]})), False)
check('A non-idempotent write is retried without an identity key', solve(dict(r, **{'idempotency_safe': [False, '']})), False)
check('A retry ignores the server requested wait period', solve(dict(r, **{'retry_after': [9, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'retryable_result': 400, 'attempt_budget': [3, 3], 'remaining_deadline': [5, 4, 8], 'idempotency_safe': [False, ''], 'retry_after': [9, 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-retry-admission

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

N = 1
observations = []
def solve(r):
    return (r['attempt_budget'][0] < r['attempt_budget'][1]) and (r['remaining_deadline'][0]+r['remaining_deadline'][1] <= r['remaining_deadline'][2]) and (r['idempotency_safe'][0] or bool(r['idempotency_safe'][1])) and (r['retry_after'][0] >= r['retry_after'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'retryable_result': 503, 'attempt_budget': [2, 3], 'remaining_deadline': [5, 3, 8], 'idempotency_safe': [False, 'op1'], 'retry_after': [10, 10]}
check('valid operation', solve(r), True)
check('A permanent rejection triggers another attempt', solve(dict(r, **{'retryable_result': 400})), False)
check('An exhausted attempt budget permits one more request', solve(dict(r, **{'attempt_budget': [3, 3]})), False)
check('A retry cannot finish within its remaining request deadline', solve(dict(r, **{'remaining_deadline': [5, 4, 8]})), False)
check('A non-idempotent write is retried without an identity key', solve(dict(r, **{'idempotency_safe': [False, '']})), False)
check('A retry ignores the server requested wait period', solve(dict(r, **{'retry_after': [9, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'retryable_result': 400, 'attempt_budget': [3, 3], 'remaining_deadline': [5, 4, 8], 'idempotency_safe': [False, ''], 'retry_after': [9, 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 retry admission operation is admitted even though a retry ignores the server requested wait period.

Return a Boolean admission decision for admit a remote operation retry. The record r must satisfy all of: r['retryable_result'] in [408,429,500,502,503,504]; r['attempt_budget'][0] < r['attempt_budget'][1]; r['remaining_deadline'][0]+r['remaining_deadline'][1] <= r['remaining_deadline'][2]; r['idempotency_safe'][0] or bool(r['idempotency_safe'][1]); r['retry_after'][0] >= r['retry_after'][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['probe_exclusive'] == 0) and (r['circuit_generation'][0] == r['circuit_generation'][1]) and (r['endpoint_health'][0] in r['endpoint_health'][1]) and (r['caller_deadline'][0] < r['caller_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'open_delay': [10, 5, 5], 'probe_exclusive': 0, 'circuit_generation': [4, 4], 'endpoint_health': ['a', ['a', 'b']], 'caller_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A probe runs before the open-state cooldown elapses', solve(dict(r, **{'open_delay': [9, 5, 5]})), False)
check('Concurrent requests occupy the single half-open probe slot', solve(dict(r, **{'probe_exclusive': 1})), False)
check('A stale timer probes a newly opened circuit generation', solve(dict(r, **{'circuit_generation': [3, 4]})), False)
check('A probe targets an endpoint removed from service discovery', solve(dict(r, **{'endpoint_health': ['c', ['a', 'b']]})), False)
check('A probe is started for an already expired caller', solve(dict(r, **{'caller_deadline': [11, 11]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'open_delay': [9, 5, 5], 'probe_exclusive': 1, 'circuit_generation': [3, 4], 'endpoint_health': ['c', ['a', 'b']], 'caller_deadline': [11, 11]}
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-circuit-probe

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

N = 1
observations = []
def solve(r):
    return (r['open_delay'][0]-r['open_delay'][1] >= r['open_delay'][2]) and (r['circuit_generation'][0] == r['circuit_generation'][1]) and (r['endpoint_health'][0] in r['endpoint_health'][1]) and (r['caller_deadline'][0] < r['caller_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'open_delay': [10, 5, 5], 'probe_exclusive': 0, 'circuit_generation': [4, 4], 'endpoint_health': ['a', ['a', 'b']], 'caller_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A probe runs before the open-state cooldown elapses', solve(dict(r, **{'open_delay': [9, 5, 5]})), False)
check('Concurrent requests occupy the single half-open probe slot', solve(dict(r, **{'probe_exclusive': 1})), False)
check('A stale timer probes a newly opened circuit generation', solve(dict(r, **{'circuit_generation': [3, 4]})), False)
check('A probe targets an endpoint removed from service discovery', solve(dict(r, **{'endpoint_health': ['c', ['a', 'b']]})), False)
check('A probe is started for an already expired caller', solve(dict(r, **{'caller_deadline': [11, 11]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'open_delay': [9, 5, 5], 'probe_exclusive': 1, 'circuit_generation': [3, 4], 'endpoint_health': ['c', ['a', 'b']], 'caller_deadline': [11, 11]}
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 circuit probe operation is admitted even though a probe runs before the open-state cooldown elapses.

Return a Boolean admission decision for allow a half-open circuit probe. The record r must satisfy all of: r['open_delay'][0]-r['open_delay'][1] >= r['open_delay'][2]; r['probe_exclusive'] == 0; r['circuit_generation'][0] == r['circuit_generation'][1]; r['endpoint_health'][0] in r['endpoint_health'][1]; r['caller_deadline'][0] < r['caller_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['open_delay'][0]-r['open_delay'][1] >= r['open_delay'][2]) and (r['circuit_generation'][0] == r['circuit_generation'][1]) and (r['endpoint_health'][0] in r['endpoint_health'][1]) and (r['caller_deadline'][0] < r['caller_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'open_delay': [10, 5, 5], 'probe_exclusive': 0, 'circuit_generation': [4, 4], 'endpoint_health': ['a', ['a', 'b']], 'caller_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A probe runs before the open-state cooldown elapses', solve(dict(r, **{'open_delay': [9, 5, 5]})), False)
check('Concurrent requests occupy the single half-open probe slot', solve(dict(r, **{'probe_exclusive': 1})), False)
check('A stale timer probes a newly opened circuit generation', solve(dict(r, **{'circuit_generation': [3, 4]})), False)
check('A probe targets an endpoint removed from service discovery', solve(dict(r, **{'endpoint_health': ['c', ['a', 'b']]})), False)
check('A probe is started for an already expired caller', solve(dict(r, **{'caller_deadline': [11, 11]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'open_delay': [9, 5, 5], 'probe_exclusive': 1, 'circuit_generation': [3, 4], 'endpoint_health': ['c', ['a', 'b']], 'caller_deadline': [11, 11]}
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-circuit-probe

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

N = 1
observations = []
def solve(r):
    return (r['open_delay'][0]-r['open_delay'][1] >= r['open_delay'][2]) and (r['probe_exclusive'] == 0) and (r['endpoint_health'][0] in r['endpoint_health'][1]) and (r['caller_deadline'][0] < r['caller_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'open_delay': [10, 5, 5], 'probe_exclusive': 0, 'circuit_generation': [4, 4], 'endpoint_health': ['a', ['a', 'b']], 'caller_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A probe runs before the open-state cooldown elapses', solve(dict(r, **{'open_delay': [9, 5, 5]})), False)
check('Concurrent requests occupy the single half-open probe slot', solve(dict(r, **{'probe_exclusive': 1})), False)
check('A stale timer probes a newly opened circuit generation', solve(dict(r, **{'circuit_generation': [3, 4]})), False)
check('A probe targets an endpoint removed from service discovery', solve(dict(r, **{'endpoint_health': ['c', ['a', 'b']]})), False)
check('A probe is started for an already expired caller', solve(dict(r, **{'caller_deadline': [11, 11]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'open_delay': [9, 5, 5], 'probe_exclusive': 1, 'circuit_generation': [3, 4], 'endpoint_health': ['c', ['a', 'b']], 'caller_deadline': [11, 11]}
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 circuit probe operation is admitted even though concurrent requests occupy the single half-open probe slot.

Return a Boolean admission decision for allow a half-open circuit probe. The record r must satisfy all of: r['open_delay'][0]-r['open_delay'][1] >= r['open_delay'][2]; r['probe_exclusive'] == 0; r['circuit_generation'][0] == r['circuit_generation'][1]; r['endpoint_health'][0] in r['endpoint_health'][1]; r['caller_deadline'][0] < r['caller_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['open_delay'][0]-r['open_delay'][1] >= r['open_delay'][2]) and (r['probe_exclusive'] == 0) and (r['endpoint_health'][0] in r['endpoint_health'][1]) and (r['caller_deadline'][0] < r['caller_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'open_delay': [10, 5, 5], 'probe_exclusive': 0, 'circuit_generation': [4, 4], 'endpoint_health': ['a', ['a', 'b']], 'caller_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A probe runs before the open-state cooldown elapses', solve(dict(r, **{'open_delay': [9, 5, 5]})), False)
check('Concurrent requests occupy the single half-open probe slot', solve(dict(r, **{'probe_exclusive': 1})), False)
check('A stale timer probes a newly opened circuit generation', solve(dict(r, **{'circuit_generation': [3, 4]})), False)
check('A probe targets an endpoint removed from service discovery', solve(dict(r, **{'endpoint_health': ['c', ['a', 'b']]})), False)
check('A probe is started for an already expired caller', solve(dict(r, **{'caller_deadline': [11, 11]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'open_delay': [9, 5, 5], 'probe_exclusive': 1, 'circuit_generation': [3, 4], 'endpoint_health': ['c', ['a', 'b']], 'caller_deadline': [11, 11]}
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-circuit-probe

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

N = 1
observations = []
def solve(r):
    return (r['open_delay'][0]-r['open_delay'][1] >= r['open_delay'][2]) and (r['probe_exclusive'] == 0) and (r['circuit_generation'][0] == r['circuit_generation'][1]) and (r['caller_deadline'][0] < r['caller_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'open_delay': [10, 5, 5], 'probe_exclusive': 0, 'circuit_generation': [4, 4], 'endpoint_health': ['a', ['a', 'b']], 'caller_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A probe runs before the open-state cooldown elapses', solve(dict(r, **{'open_delay': [9, 5, 5]})), False)
check('Concurrent requests occupy the single half-open probe slot', solve(dict(r, **{'probe_exclusive': 1})), False)
check('A stale timer probes a newly opened circuit generation', solve(dict(r, **{'circuit_generation': [3, 4]})), False)
check('A probe targets an endpoint removed from service discovery', solve(dict(r, **{'endpoint_health': ['c', ['a', 'b']]})), False)
check('A probe is started for an already expired caller', solve(dict(r, **{'caller_deadline': [11, 11]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'open_delay': [9, 5, 5], 'probe_exclusive': 1, 'circuit_generation': [3, 4], 'endpoint_health': ['c', ['a', 'b']], 'caller_deadline': [11, 11]}
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 circuit probe operation is admitted even though a stale timer probes a newly opened circuit generation.

Return a Boolean admission decision for allow a half-open circuit probe. The record r must satisfy all of: r['open_delay'][0]-r['open_delay'][1] >= r['open_delay'][2]; r['probe_exclusive'] == 0; r['circuit_generation'][0] == r['circuit_generation'][1]; r['endpoint_health'][0] in r['endpoint_health'][1]; r['caller_deadline'][0] < r['caller_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['open_delay'][0]-r['open_delay'][1] >= r['open_delay'][2]) and (r['probe_exclusive'] == 0) and (r['circuit_generation'][0] == r['circuit_generation'][1]) and (r['caller_deadline'][0] < r['caller_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'open_delay': [10, 5, 5], 'probe_exclusive': 0, 'circuit_generation': [4, 4], 'endpoint_health': ['a', ['a', 'b']], 'caller_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A probe runs before the open-state cooldown elapses', solve(dict(r, **{'open_delay': [9, 5, 5]})), False)
check('Concurrent requests occupy the single half-open probe slot', solve(dict(r, **{'probe_exclusive': 1})), False)
check('A stale timer probes a newly opened circuit generation', solve(dict(r, **{'circuit_generation': [3, 4]})), False)
check('A probe targets an endpoint removed from service discovery', solve(dict(r, **{'endpoint_health': ['c', ['a', 'b']]})), False)
check('A probe is started for an already expired caller', solve(dict(r, **{'caller_deadline': [11, 11]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'open_delay': [9, 5, 5], 'probe_exclusive': 1, 'circuit_generation': [3, 4], 'endpoint_health': ['c', ['a', 'b']], 'caller_deadline': [11, 11]}
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-circuit-probe

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

N = 1
observations = []
def solve(r):
    return (r['open_delay'][0]-r['open_delay'][1] >= r['open_delay'][2]) and (r['probe_exclusive'] == 0) and (r['circuit_generation'][0] == r['circuit_generation'][1]) and (r['endpoint_health'][0] in r['endpoint_health'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'open_delay': [10, 5, 5], 'probe_exclusive': 0, 'circuit_generation': [4, 4], 'endpoint_health': ['a', ['a', 'b']], 'caller_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A probe runs before the open-state cooldown elapses', solve(dict(r, **{'open_delay': [9, 5, 5]})), False)
check('Concurrent requests occupy the single half-open probe slot', solve(dict(r, **{'probe_exclusive': 1})), False)
check('A stale timer probes a newly opened circuit generation', solve(dict(r, **{'circuit_generation': [3, 4]})), False)
check('A probe targets an endpoint removed from service discovery', solve(dict(r, **{'endpoint_health': ['c', ['a', 'b']]})), False)
check('A probe is started for an already expired caller', solve(dict(r, **{'caller_deadline': [11, 11]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'open_delay': [9, 5, 5], 'probe_exclusive': 1, 'circuit_generation': [3, 4], 'endpoint_health': ['c', ['a', 'b']], 'caller_deadline': [11, 11]}
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 circuit probe operation is admitted even though a probe targets an endpoint removed from service discovery.

Return a Boolean admission decision for allow a half-open circuit probe. The record r must satisfy all of: r['open_delay'][0]-r['open_delay'][1] >= r['open_delay'][2]; r['probe_exclusive'] == 0; r['circuit_generation'][0] == r['circuit_generation'][1]; r['endpoint_health'][0] in r['endpoint_health'][1]; r['caller_deadline'][0] < r['caller_deadline'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access