broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json
from fractions import Fraction
import math
N = 1
observations = []
def solve(numerator, denominator):
    return round(Fraction(numerator, denominator)) if denominator > 0 else None
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('positive half after even integer', solve(4 * N + 1, 2), 2 * N + 1)
check('negative half after even magnitude', solve(-4 * N - 1, 2), -2 * N - 1)
check('just below midpoint', solve(200 * N + 49, 100), 2 * N)
check('just above midpoint', solve(200 * N + 51, 100), 2 * N + 1)
check('negative value below midpoint magnitude', solve(-200 * N - 49, 100), -2 * N)
check('exact negative integer', solve(-N * 7, 7), -N)
check('zero', solve(0, N), 0)
check('zero denominator rejected', solve(N, 0), None)
check('negative denominator rejected', solve(N, -1), None)
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)

source:
"""Failure Map reference implementation. Python standard library only."""
import json
from fractions import Fraction
import math
N = 1
observations = []
def solve(numerator, denominator):
    return math.floor(Fraction(numerator, denominator) + Fraction(1, 2)) if denominator > 0 else None
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('positive half after even integer', solve(4 * N + 1, 2), 2 * N + 1)
check('negative half after even magnitude', solve(-4 * N - 1, 2), -2 * N - 1)
check('just below midpoint', solve(200 * N + 49, 100), 2 * N)
check('just above midpoint', solve(200 * N + 51, 100), 2 * N + 1)
check('negative value below midpoint magnitude', solve(-200 * N - 49, 100), -2 * N)
check('exact negative integer', solve(-N * 7, 7), -N)
check('zero', solve(0, N), 0)
check('zero denominator rejected', solve(N, 0), None)
check('negative denominator rejected', solve(N, -1), None)
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:
Exact midpoint values receive even rounding or asymmetric rounding instead of the required away-from-zero result.

Round numerator/denominator to the nearest integer, with exact ties away from zero. Inputs are integers and denominator must be positive; invalid denominators return None. No decimal parsing or floating-point conversion occurs.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json
import math
import sys
N = 1
observations = []
def solve(values):
    if not values or any(value <= 0 for value in values):
        return None
    return format(math.prod(values) ** (1 / len(values)), '.10g')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('finite mean despite overflowing product', solve([1e150] * (N + 2)), '1e+150')
check('finite mean despite underflowing product', solve([1e-200] * (N + 1)), '1e-200')
check('balanced reciprocal scales', solve([1e300, 1e-300] * N), '1')
check('ordinary square-root case', solve([4.0, 9.0]), '6')
check('one observation', solve([float(N)]), str(N))
check('unit-valued observations', solve([1.0] * (N + 1)), '1')
check('zero observation rejected', solve([0.0, 1.0]), None)
check('negative observation rejected', solve([-1.0, 2.0]), None)
check('empty sequence', solve([]), None)
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

source:
"""Failure Map reference implementation. Python standard library only."""
import json
import math
import sys
N = 1
observations = []
def solve(values):
    if not values or any(value <= 0 for value in values):
        return None
    product = 1.0
    for value in values:
        product = min(product * value, sys.float_info.max)
    return format(product ** (1 / len(values)), '.10g')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('finite mean despite overflowing product', solve([1e150] * (N + 2)), '1e+150')
check('finite mean despite underflowing product', solve([1e-200] * (N + 1)), '1e-200')
check('balanced reciprocal scales', solve([1e300, 1e-300] * N), '1')
check('ordinary square-root case', solve([4.0, 9.0]), '6')
check('one observation', solve([float(N)]), str(N))
check('unit-valued observations', solve([1.0] * (N + 1)), '1')
check('zero observation rejected', solve([0.0, 1.0]), None)
check('negative observation rejected', solve([-1.0, 2.0]), None)
check('empty sequence', solve([]), None)
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:
Multiplying observations produces infinity or zero even when their geometric mean is a representable positive number.

For a nonempty sequence of finite positive floats, return a logarithm-based geometric mean formatted with ten significant digits. Empty input or any nonpositive observation returns None. Fixtures remain within the representable range of the final mean.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (len(set(r['new_majority'][0]) & set(r['new_majority'][1])) * 2 > len(set(r['new_majority'][0]))) and (r['configuration_entry'][0] <= r['configuration_entry'][1]) and (r['no_pending_change'] is None) and (all(x >= r['learner_catchup'][0] for x in r['learner_catchup'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'old_majority': [[1, 2, 3], [1, 2]], 'new_majority': [[3, 4, 5], [3, 4]], 'configuration_entry': [8, 8], 'no_pending_change': None, 'learner_catchup': [8, [8, 10]]}
check('valid operation', solve(r), True)
check('The old configuration has only a minority', solve(dict(r, **{'old_majority': [[1, 2, 3], [1]]})), False)
check('The new configuration has only a minority', solve(dict(r, **{'new_majority': [[3, 4, 5], [3]]})), False)
check('The configuration entry is beyond the committed prefix', solve(dict(r, **{'configuration_entry': [9, 8]})), False)
check('A second membership change overlaps an unfinalized change', solve(dict(r, **{'no_pending_change': 'pending'})), False)
check('A promoted learner lacks the configuration entry', solve(dict(r, **{'learner_catchup': [8, [7, 10]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'old_majority': [[1, 2, 3], [1]], 'new_majority': [[3, 4, 5], [3]], 'configuration_entry': [9, 8], 'no_pending_change': 'pending', 'learner_catchup': [8, [7, 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-joint-configuration

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

N = 1
observations = []
def solve(r):
    return (len(set(r['old_majority'][0]) & set(r['old_majority'][1])) * 2 > len(set(r['old_majority'][0]))) and (r['configuration_entry'][0] <= r['configuration_entry'][1]) and (r['no_pending_change'] is None) and (all(x >= r['learner_catchup'][0] for x in r['learner_catchup'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'old_majority': [[1, 2, 3], [1, 2]], 'new_majority': [[3, 4, 5], [3, 4]], 'configuration_entry': [8, 8], 'no_pending_change': None, 'learner_catchup': [8, [8, 10]]}
check('valid operation', solve(r), True)
check('The old configuration has only a minority', solve(dict(r, **{'old_majority': [[1, 2, 3], [1]]})), False)
check('The new configuration has only a minority', solve(dict(r, **{'new_majority': [[3, 4, 5], [3]]})), False)
check('The configuration entry is beyond the committed prefix', solve(dict(r, **{'configuration_entry': [9, 8]})), False)
check('A second membership change overlaps an unfinalized change', solve(dict(r, **{'no_pending_change': 'pending'})), False)
check('A promoted learner lacks the configuration entry', solve(dict(r, **{'learner_catchup': [8, [7, 10]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'old_majority': [[1, 2, 3], [1]], 'new_majority': [[3, 4, 5], [3]], 'configuration_entry': [9, 8], 'no_pending_change': 'pending', 'learner_catchup': [8, [7, 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 joint configuration operation is admitted even though the old configuration has only a minority.

Return a Boolean admission decision for activate a joint voting configuration. The record r must satisfy all of: len(set(r['old_majority'][0]) & set(r['old_majority'][1])) * 2 > len(set(r['old_majority'][0])); len(set(r['new_majority'][0]) & set(r['new_majority'][1])) * 2 > len(set(r['new_majority'][0])); r['configuration_entry'][0] <= r['configuration_entry'][1]; r['no_pending_change'] is None; all(x >= r['learner_catchup'][0] for x in r['learner_catchup'][1]). Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (len(set(r['old_majority'][0]) & set(r['old_majority'][1])) * 2 > len(set(r['old_majority'][0]))) and (r['configuration_entry'][0] <= r['configuration_entry'][1]) and (r['no_pending_change'] is None) and (all(x >= r['learner_catchup'][0] for x in r['learner_catchup'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'old_majority': [[1, 2, 3], [1, 2]], 'new_majority': [[3, 4, 5], [3, 4]], 'configuration_entry': [8, 8], 'no_pending_change': None, 'learner_catchup': [8, [8, 10]]}
check('valid operation', solve(r), True)
check('The old configuration has only a minority', solve(dict(r, **{'old_majority': [[1, 2, 3], [1]]})), False)
check('The new configuration has only a minority', solve(dict(r, **{'new_majority': [[3, 4, 5], [3]]})), False)
check('The configuration entry is beyond the committed prefix', solve(dict(r, **{'configuration_entry': [9, 8]})), False)
check('A second membership change overlaps an unfinalized change', solve(dict(r, **{'no_pending_change': 'pending'})), False)
check('A promoted learner lacks the configuration entry', solve(dict(r, **{'learner_catchup': [8, [7, 10]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'old_majority': [[1, 2, 3], [1]], 'new_majority': [[3, 4, 5], [3]], 'configuration_entry': [9, 8], 'no_pending_change': 'pending', 'learner_catchup': [8, [7, 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-joint-configuration

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

N = 1
observations = []
def solve(r):
    return (len(set(r['old_majority'][0]) & set(r['old_majority'][1])) * 2 > len(set(r['old_majority'][0]))) and (len(set(r['new_majority'][0]) & set(r['new_majority'][1])) * 2 > len(set(r['new_majority'][0]))) and (r['no_pending_change'] is None) and (all(x >= r['learner_catchup'][0] for x in r['learner_catchup'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'old_majority': [[1, 2, 3], [1, 2]], 'new_majority': [[3, 4, 5], [3, 4]], 'configuration_entry': [8, 8], 'no_pending_change': None, 'learner_catchup': [8, [8, 10]]}
check('valid operation', solve(r), True)
check('The old configuration has only a minority', solve(dict(r, **{'old_majority': [[1, 2, 3], [1]]})), False)
check('The new configuration has only a minority', solve(dict(r, **{'new_majority': [[3, 4, 5], [3]]})), False)
check('The configuration entry is beyond the committed prefix', solve(dict(r, **{'configuration_entry': [9, 8]})), False)
check('A second membership change overlaps an unfinalized change', solve(dict(r, **{'no_pending_change': 'pending'})), False)
check('A promoted learner lacks the configuration entry', solve(dict(r, **{'learner_catchup': [8, [7, 10]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'old_majority': [[1, 2, 3], [1]], 'new_majority': [[3, 4, 5], [3]], 'configuration_entry': [9, 8], 'no_pending_change': 'pending', 'learner_catchup': [8, [7, 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 joint configuration operation is admitted even though the new configuration has only a minority.

Return a Boolean admission decision for activate a joint voting configuration. The record r must satisfy all of: len(set(r['old_majority'][0]) & set(r['old_majority'][1])) * 2 > len(set(r['old_majority'][0])); len(set(r['new_majority'][0]) & set(r['new_majority'][1])) * 2 > len(set(r['new_majority'][0])); r['configuration_entry'][0] <= r['configuration_entry'][1]; r['no_pending_change'] is None; all(x >= r['learner_catchup'][0] for x in r['learner_catchup'][1]). Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (len(set(r['old_majority'][0]) & set(r['old_majority'][1])) * 2 > len(set(r['old_majority'][0]))) and (len(set(r['new_majority'][0]) & set(r['new_majority'][1])) * 2 > len(set(r['new_majority'][0]))) and (r['no_pending_change'] is None) and (all(x >= r['learner_catchup'][0] for x in r['learner_catchup'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'old_majority': [[1, 2, 3], [1, 2]], 'new_majority': [[3, 4, 5], [3, 4]], 'configuration_entry': [8, 8], 'no_pending_change': None, 'learner_catchup': [8, [8, 10]]}
check('valid operation', solve(r), True)
check('The old configuration has only a minority', solve(dict(r, **{'old_majority': [[1, 2, 3], [1]]})), False)
check('The new configuration has only a minority', solve(dict(r, **{'new_majority': [[3, 4, 5], [3]]})), False)
check('The configuration entry is beyond the committed prefix', solve(dict(r, **{'configuration_entry': [9, 8]})), False)
check('A second membership change overlaps an unfinalized change', solve(dict(r, **{'no_pending_change': 'pending'})), False)
check('A promoted learner lacks the configuration entry', solve(dict(r, **{'learner_catchup': [8, [7, 10]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'old_majority': [[1, 2, 3], [1]], 'new_majority': [[3, 4, 5], [3]], 'configuration_entry': [9, 8], 'no_pending_change': 'pending', 'learner_catchup': [8, [7, 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-joint-configuration

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

N = 1
observations = []
def solve(r):
    return (len(set(r['old_majority'][0]) & set(r['old_majority'][1])) * 2 > len(set(r['old_majority'][0]))) and (len(set(r['new_majority'][0]) & set(r['new_majority'][1])) * 2 > len(set(r['new_majority'][0]))) and (r['configuration_entry'][0] <= r['configuration_entry'][1]) and (all(x >= r['learner_catchup'][0] for x in r['learner_catchup'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'old_majority': [[1, 2, 3], [1, 2]], 'new_majority': [[3, 4, 5], [3, 4]], 'configuration_entry': [8, 8], 'no_pending_change': None, 'learner_catchup': [8, [8, 10]]}
check('valid operation', solve(r), True)
check('The old configuration has only a minority', solve(dict(r, **{'old_majority': [[1, 2, 3], [1]]})), False)
check('The new configuration has only a minority', solve(dict(r, **{'new_majority': [[3, 4, 5], [3]]})), False)
check('The configuration entry is beyond the committed prefix', solve(dict(r, **{'configuration_entry': [9, 8]})), False)
check('A second membership change overlaps an unfinalized change', solve(dict(r, **{'no_pending_change': 'pending'})), False)
check('A promoted learner lacks the configuration entry', solve(dict(r, **{'learner_catchup': [8, [7, 10]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'old_majority': [[1, 2, 3], [1]], 'new_majority': [[3, 4, 5], [3]], 'configuration_entry': [9, 8], 'no_pending_change': 'pending', 'learner_catchup': [8, [7, 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 joint configuration operation is admitted even though the configuration entry is beyond the committed prefix.

Return a Boolean admission decision for activate a joint voting configuration. The record r must satisfy all of: len(set(r['old_majority'][0]) & set(r['old_majority'][1])) * 2 > len(set(r['old_majority'][0])); len(set(r['new_majority'][0]) & set(r['new_majority'][1])) * 2 > len(set(r['new_majority'][0])); r['configuration_entry'][0] <= r['configuration_entry'][1]; r['no_pending_change'] is None; all(x >= r['learner_catchup'][0] for x in r['learner_catchup'][1]). Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (len(set(r['old_majority'][0]) & set(r['old_majority'][1])) * 2 > len(set(r['old_majority'][0]))) and (len(set(r['new_majority'][0]) & set(r['new_majority'][1])) * 2 > len(set(r['new_majority'][0]))) and (r['configuration_entry'][0] <= r['configuration_entry'][1]) and (all(x >= r['learner_catchup'][0] for x in r['learner_catchup'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'old_majority': [[1, 2, 3], [1, 2]], 'new_majority': [[3, 4, 5], [3, 4]], 'configuration_entry': [8, 8], 'no_pending_change': None, 'learner_catchup': [8, [8, 10]]}
check('valid operation', solve(r), True)
check('The old configuration has only a minority', solve(dict(r, **{'old_majority': [[1, 2, 3], [1]]})), False)
check('The new configuration has only a minority', solve(dict(r, **{'new_majority': [[3, 4, 5], [3]]})), False)
check('The configuration entry is beyond the committed prefix', solve(dict(r, **{'configuration_entry': [9, 8]})), False)
check('A second membership change overlaps an unfinalized change', solve(dict(r, **{'no_pending_change': 'pending'})), False)
check('A promoted learner lacks the configuration entry', solve(dict(r, **{'learner_catchup': [8, [7, 10]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'old_majority': [[1, 2, 3], [1]], 'new_majority': [[3, 4, 5], [3]], 'configuration_entry': [9, 8], 'no_pending_change': 'pending', 'learner_catchup': [8, [7, 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-joint-configuration

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

N = 1
observations = []
def solve(r):
    return (len(set(r['old_majority'][0]) & set(r['old_majority'][1])) * 2 > len(set(r['old_majority'][0]))) and (len(set(r['new_majority'][0]) & set(r['new_majority'][1])) * 2 > len(set(r['new_majority'][0]))) and (r['configuration_entry'][0] <= r['configuration_entry'][1]) and (r['no_pending_change'] is None)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'old_majority': [[1, 2, 3], [1, 2]], 'new_majority': [[3, 4, 5], [3, 4]], 'configuration_entry': [8, 8], 'no_pending_change': None, 'learner_catchup': [8, [8, 10]]}
check('valid operation', solve(r), True)
check('The old configuration has only a minority', solve(dict(r, **{'old_majority': [[1, 2, 3], [1]]})), False)
check('The new configuration has only a minority', solve(dict(r, **{'new_majority': [[3, 4, 5], [3]]})), False)
check('The configuration entry is beyond the committed prefix', solve(dict(r, **{'configuration_entry': [9, 8]})), False)
check('A second membership change overlaps an unfinalized change', solve(dict(r, **{'no_pending_change': 'pending'})), False)
check('A promoted learner lacks the configuration entry', solve(dict(r, **{'learner_catchup': [8, [7, 10]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'old_majority': [[1, 2, 3], [1]], 'new_majority': [[3, 4, 5], [3]], 'configuration_entry': [9, 8], 'no_pending_change': 'pending', 'learner_catchup': [8, [7, 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 joint configuration operation is admitted even though a second membership change overlaps an unfinalized change.

Return a Boolean admission decision for activate a joint voting configuration. The record r must satisfy all of: len(set(r['old_majority'][0]) & set(r['old_majority'][1])) * 2 > len(set(r['old_majority'][0])); len(set(r['new_majority'][0]) & set(r['new_majority'][1])) * 2 > len(set(r['new_majority'][0])); r['configuration_entry'][0] <= r['configuration_entry'][1]; r['no_pending_change'] is None; all(x >= r['learner_catchup'][0] for x in r['learner_catchup'][1]). Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (len(set(r['old_majority'][0]) & set(r['old_majority'][1])) * 2 > len(set(r['old_majority'][0]))) and (len(set(r['new_majority'][0]) & set(r['new_majority'][1])) * 2 > len(set(r['new_majority'][0]))) and (r['configuration_entry'][0] <= r['configuration_entry'][1]) and (r['no_pending_change'] is None)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'old_majority': [[1, 2, 3], [1, 2]], 'new_majority': [[3, 4, 5], [3, 4]], 'configuration_entry': [8, 8], 'no_pending_change': None, 'learner_catchup': [8, [8, 10]]}
check('valid operation', solve(r), True)
check('The old configuration has only a minority', solve(dict(r, **{'old_majority': [[1, 2, 3], [1]]})), False)
check('The new configuration has only a minority', solve(dict(r, **{'new_majority': [[3, 4, 5], [3]]})), False)
check('The configuration entry is beyond the committed prefix', solve(dict(r, **{'configuration_entry': [9, 8]})), False)
check('A second membership change overlaps an unfinalized change', solve(dict(r, **{'no_pending_change': 'pending'})), False)
check('A promoted learner lacks the configuration entry', solve(dict(r, **{'learner_catchup': [8, [7, 10]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'old_majority': [[1, 2, 3], [1]], 'new_majority': [[3, 4, 5], [3]], 'configuration_entry': [9, 8], 'no_pending_change': 'pending', 'learner_catchup': [8, [7, 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-joint-configuration

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

N = 1
observations = []
def solve(r):
    return (len(set(r['new_majority'][0]) & set(r['new_majority'][1])) * 2 > len(set(r['new_majority'][0]))) and (r['configuration_entry'][0] <= r['configuration_entry'][1]) and (r['no_pending_change'] is None) and (all(x >= r['learner_catchup'][0] for x in r['learner_catchup'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'old_majority': [[1, 2, 3], [1, 2]], 'new_majority': [[3, 4, 5], [3, 4]], 'configuration_entry': [8, 8], 'no_pending_change': None, 'learner_catchup': [8, [8, 10]]}
check('valid operation', solve(r), True)
check('The old configuration has only a minority', solve(dict(r, **{'old_majority': [[1, 2, 3], [1]]})), False)
check('The new configuration has only a minority', solve(dict(r, **{'new_majority': [[3, 4, 5], [3]]})), False)
check('The configuration entry is beyond the committed prefix', solve(dict(r, **{'configuration_entry': [9, 8]})), False)
check('A second membership change overlaps an unfinalized change', solve(dict(r, **{'no_pending_change': 'pending'})), False)
check('A promoted learner lacks the configuration entry', solve(dict(r, **{'learner_catchup': [8, [7, 10]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'old_majority': [[1, 2, 3], [1]], 'new_majority': [[3, 4, 5], [3]], 'configuration_entry': [9, 8], 'no_pending_change': 'pending', 'learner_catchup': [8, [7, 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 joint configuration operation is admitted even though a promoted learner lacks the configuration entry.

Return a Boolean admission decision for activate a joint voting configuration. The record r must satisfy all of: len(set(r['old_majority'][0]) & set(r['old_majority'][1])) * 2 > len(set(r['old_majority'][0])); len(set(r['new_majority'][0]) & set(r['new_majority'][1])) * 2 > len(set(r['new_majority'][0])); r['configuration_entry'][0] <= r['configuration_entry'][1]; r['no_pending_change'] is None; all(x >= r['learner_catchup'][0] for x in r['learner_catchup'][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['log_caught_up'][0] >= r['log_caught_up'][1]) and (r['transfer_term'][0] == r['transfer_term'][1]) and (r['transfer_deadline'][0] < r['transfer_deadline'][1]) and (r['writes_drained'] == 0)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'target_voter': ['b', ['a', 'b']], 'log_caught_up': [12, 12], 'transfer_term': [4, 4], 'transfer_deadline': [9, 10], 'writes_drained': 0}
check('valid operation', solve(r), True)
check('Leadership is handed to a learner', solve(dict(r, **{'target_voter': ['c', ['a', 'b']]})), False)
check('The target lacks the leader final log entry', solve(dict(r, **{'log_caught_up': [11, 12]})), False)
check('A delayed handoff names an obsolete term', solve(dict(r, **{'transfer_term': [3, 4]})), False)
check('The handoff arrives after its transfer deadline', solve(dict(r, **{'transfer_deadline': [10, 10]})), False)
check('A transfer races outstanding leader writes', solve(dict(r, **{'writes_drained': 1})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'target_voter': ['c', ['a', 'b']], 'log_caught_up': [11, 12], 'transfer_term': [3, 4], 'transfer_deadline': [10, 10], 'writes_drained': 1}
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-leader-transfer

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

N = 1
observations = []
def solve(r):
    return (r['target_voter'][0] in r['target_voter'][1]) and (r['transfer_term'][0] == r['transfer_term'][1]) and (r['transfer_deadline'][0] < r['transfer_deadline'][1]) and (r['writes_drained'] == 0)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'target_voter': ['b', ['a', 'b']], 'log_caught_up': [12, 12], 'transfer_term': [4, 4], 'transfer_deadline': [9, 10], 'writes_drained': 0}
check('valid operation', solve(r), True)
check('Leadership is handed to a learner', solve(dict(r, **{'target_voter': ['c', ['a', 'b']]})), False)
check('The target lacks the leader final log entry', solve(dict(r, **{'log_caught_up': [11, 12]})), False)
check('A delayed handoff names an obsolete term', solve(dict(r, **{'transfer_term': [3, 4]})), False)
check('The handoff arrives after its transfer deadline', solve(dict(r, **{'transfer_deadline': [10, 10]})), False)
check('A transfer races outstanding leader writes', solve(dict(r, **{'writes_drained': 1})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'target_voter': ['c', ['a', 'b']], 'log_caught_up': [11, 12], 'transfer_term': [3, 4], 'transfer_deadline': [10, 10], 'writes_drained': 1}
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 leader transfer operation is admitted even though leadership is handed to a learner.

Return a Boolean admission decision for transfer leadership to a chosen voter. The record r must satisfy all of: r['target_voter'][0] in r['target_voter'][1]; r['log_caught_up'][0] >= r['log_caught_up'][1]; r['transfer_term'][0] == r['transfer_term'][1]; r['transfer_deadline'][0] < r['transfer_deadline'][1]; r['writes_drained'] == 0. 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['target_voter'][0] in r['target_voter'][1]) and (r['transfer_term'][0] == r['transfer_term'][1]) and (r['transfer_deadline'][0] < r['transfer_deadline'][1]) and (r['writes_drained'] == 0)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'target_voter': ['b', ['a', 'b']], 'log_caught_up': [12, 12], 'transfer_term': [4, 4], 'transfer_deadline': [9, 10], 'writes_drained': 0}
check('valid operation', solve(r), True)
check('Leadership is handed to a learner', solve(dict(r, **{'target_voter': ['c', ['a', 'b']]})), False)
check('The target lacks the leader final log entry', solve(dict(r, **{'log_caught_up': [11, 12]})), False)
check('A delayed handoff names an obsolete term', solve(dict(r, **{'transfer_term': [3, 4]})), False)
check('The handoff arrives after its transfer deadline', solve(dict(r, **{'transfer_deadline': [10, 10]})), False)
check('A transfer races outstanding leader writes', solve(dict(r, **{'writes_drained': 1})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'target_voter': ['c', ['a', 'b']], 'log_caught_up': [11, 12], 'transfer_term': [3, 4], 'transfer_deadline': [10, 10], 'writes_drained': 1}
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-leader-transfer

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

N = 1
observations = []
def solve(r):
    return (r['target_voter'][0] in r['target_voter'][1]) and (r['log_caught_up'][0] >= r['log_caught_up'][1]) and (r['transfer_deadline'][0] < r['transfer_deadline'][1]) and (r['writes_drained'] == 0)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'target_voter': ['b', ['a', 'b']], 'log_caught_up': [12, 12], 'transfer_term': [4, 4], 'transfer_deadline': [9, 10], 'writes_drained': 0}
check('valid operation', solve(r), True)
check('Leadership is handed to a learner', solve(dict(r, **{'target_voter': ['c', ['a', 'b']]})), False)
check('The target lacks the leader final log entry', solve(dict(r, **{'log_caught_up': [11, 12]})), False)
check('A delayed handoff names an obsolete term', solve(dict(r, **{'transfer_term': [3, 4]})), False)
check('The handoff arrives after its transfer deadline', solve(dict(r, **{'transfer_deadline': [10, 10]})), False)
check('A transfer races outstanding leader writes', solve(dict(r, **{'writes_drained': 1})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'target_voter': ['c', ['a', 'b']], 'log_caught_up': [11, 12], 'transfer_term': [3, 4], 'transfer_deadline': [10, 10], 'writes_drained': 1}
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 leader transfer operation is admitted even though the target lacks the leader final log entry.

Return a Boolean admission decision for transfer leadership to a chosen voter. The record r must satisfy all of: r['target_voter'][0] in r['target_voter'][1]; r['log_caught_up'][0] >= r['log_caught_up'][1]; r['transfer_term'][0] == r['transfer_term'][1]; r['transfer_deadline'][0] < r['transfer_deadline'][1]; r['writes_drained'] == 0. 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['target_voter'][0] in r['target_voter'][1]) and (r['log_caught_up'][0] >= r['log_caught_up'][1]) and (r['transfer_deadline'][0] < r['transfer_deadline'][1]) and (r['writes_drained'] == 0)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'target_voter': ['b', ['a', 'b']], 'log_caught_up': [12, 12], 'transfer_term': [4, 4], 'transfer_deadline': [9, 10], 'writes_drained': 0}
check('valid operation', solve(r), True)
check('Leadership is handed to a learner', solve(dict(r, **{'target_voter': ['c', ['a', 'b']]})), False)
check('The target lacks the leader final log entry', solve(dict(r, **{'log_caught_up': [11, 12]})), False)
check('A delayed handoff names an obsolete term', solve(dict(r, **{'transfer_term': [3, 4]})), False)
check('The handoff arrives after its transfer deadline', solve(dict(r, **{'transfer_deadline': [10, 10]})), False)
check('A transfer races outstanding leader writes', solve(dict(r, **{'writes_drained': 1})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'target_voter': ['c', ['a', 'b']], 'log_caught_up': [11, 12], 'transfer_term': [3, 4], 'transfer_deadline': [10, 10], 'writes_drained': 1}
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-leader-transfer

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

N = 1
observations = []
def solve(r):
    return (r['target_voter'][0] in r['target_voter'][1]) and (r['log_caught_up'][0] >= r['log_caught_up'][1]) and (r['transfer_term'][0] == r['transfer_term'][1]) and (r['writes_drained'] == 0)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'target_voter': ['b', ['a', 'b']], 'log_caught_up': [12, 12], 'transfer_term': [4, 4], 'transfer_deadline': [9, 10], 'writes_drained': 0}
check('valid operation', solve(r), True)
check('Leadership is handed to a learner', solve(dict(r, **{'target_voter': ['c', ['a', 'b']]})), False)
check('The target lacks the leader final log entry', solve(dict(r, **{'log_caught_up': [11, 12]})), False)
check('A delayed handoff names an obsolete term', solve(dict(r, **{'transfer_term': [3, 4]})), False)
check('The handoff arrives after its transfer deadline', solve(dict(r, **{'transfer_deadline': [10, 10]})), False)
check('A transfer races outstanding leader writes', solve(dict(r, **{'writes_drained': 1})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'target_voter': ['c', ['a', 'b']], 'log_caught_up': [11, 12], 'transfer_term': [3, 4], 'transfer_deadline': [10, 10], 'writes_drained': 1}
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 leader transfer operation is admitted even though a delayed handoff names an obsolete term.

Return a Boolean admission decision for transfer leadership to a chosen voter. The record r must satisfy all of: r['target_voter'][0] in r['target_voter'][1]; r['log_caught_up'][0] >= r['log_caught_up'][1]; r['transfer_term'][0] == r['transfer_term'][1]; r['transfer_deadline'][0] < r['transfer_deadline'][1]; r['writes_drained'] == 0. 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['target_voter'][0] in r['target_voter'][1]) and (r['log_caught_up'][0] >= r['log_caught_up'][1]) and (r['transfer_term'][0] == r['transfer_term'][1]) and (r['writes_drained'] == 0)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'target_voter': ['b', ['a', 'b']], 'log_caught_up': [12, 12], 'transfer_term': [4, 4], 'transfer_deadline': [9, 10], 'writes_drained': 0}
check('valid operation', solve(r), True)
check('Leadership is handed to a learner', solve(dict(r, **{'target_voter': ['c', ['a', 'b']]})), False)
check('The target lacks the leader final log entry', solve(dict(r, **{'log_caught_up': [11, 12]})), False)
check('A delayed handoff names an obsolete term', solve(dict(r, **{'transfer_term': [3, 4]})), False)
check('The handoff arrives after its transfer deadline', solve(dict(r, **{'transfer_deadline': [10, 10]})), False)
check('A transfer races outstanding leader writes', solve(dict(r, **{'writes_drained': 1})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'target_voter': ['c', ['a', 'b']], 'log_caught_up': [11, 12], 'transfer_term': [3, 4], 'transfer_deadline': [10, 10], 'writes_drained': 1}
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-leader-transfer

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

N = 1
observations = []
def solve(r):
    return (r['target_voter'][0] in r['target_voter'][1]) and (r['log_caught_up'][0] >= r['log_caught_up'][1]) and (r['transfer_term'][0] == r['transfer_term'][1]) and (r['transfer_deadline'][0] < r['transfer_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'target_voter': ['b', ['a', 'b']], 'log_caught_up': [12, 12], 'transfer_term': [4, 4], 'transfer_deadline': [9, 10], 'writes_drained': 0}
check('valid operation', solve(r), True)
check('Leadership is handed to a learner', solve(dict(r, **{'target_voter': ['c', ['a', 'b']]})), False)
check('The target lacks the leader final log entry', solve(dict(r, **{'log_caught_up': [11, 12]})), False)
check('A delayed handoff names an obsolete term', solve(dict(r, **{'transfer_term': [3, 4]})), False)
check('The handoff arrives after its transfer deadline', solve(dict(r, **{'transfer_deadline': [10, 10]})), False)
check('A transfer races outstanding leader writes', solve(dict(r, **{'writes_drained': 1})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'target_voter': ['c', ['a', 'b']], 'log_caught_up': [11, 12], 'transfer_term': [3, 4], 'transfer_deadline': [10, 10], 'writes_drained': 1}
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 leader transfer operation is admitted even though the handoff arrives after its transfer deadline.

Return a Boolean admission decision for transfer leadership to a chosen voter. The record r must satisfy all of: r['target_voter'][0] in r['target_voter'][1]; r['log_caught_up'][0] >= r['log_caught_up'][1]; r['transfer_term'][0] == r['transfer_term'][1]; r['transfer_deadline'][0] < r['transfer_deadline'][1]; r['writes_drained'] == 0. 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['target_voter'][0] in r['target_voter'][1]) and (r['log_caught_up'][0] >= r['log_caught_up'][1]) and (r['transfer_term'][0] == r['transfer_term'][1]) and (r['transfer_deadline'][0] < r['transfer_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'target_voter': ['b', ['a', 'b']], 'log_caught_up': [12, 12], 'transfer_term': [4, 4], 'transfer_deadline': [9, 10], 'writes_drained': 0}
check('valid operation', solve(r), True)
check('Leadership is handed to a learner', solve(dict(r, **{'target_voter': ['c', ['a', 'b']]})), False)
check('The target lacks the leader final log entry', solve(dict(r, **{'log_caught_up': [11, 12]})), False)
check('A delayed handoff names an obsolete term', solve(dict(r, **{'transfer_term': [3, 4]})), False)
check('The handoff arrives after its transfer deadline', solve(dict(r, **{'transfer_deadline': [10, 10]})), False)
check('A transfer races outstanding leader writes', solve(dict(r, **{'writes_drained': 1})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'target_voter': ['c', ['a', 'b']], 'log_caught_up': [11, 12], 'transfer_term': [3, 4], 'transfer_deadline': [10, 10], 'writes_drained': 1}
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-leader-transfer

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

N = 1
observations = []
def solve(r):
    return (r['log_caught_up'][0] >= r['log_caught_up'][1]) and (r['transfer_term'][0] == r['transfer_term'][1]) and (r['transfer_deadline'][0] < r['transfer_deadline'][1]) and (r['writes_drained'] == 0)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'target_voter': ['b', ['a', 'b']], 'log_caught_up': [12, 12], 'transfer_term': [4, 4], 'transfer_deadline': [9, 10], 'writes_drained': 0}
check('valid operation', solve(r), True)
check('Leadership is handed to a learner', solve(dict(r, **{'target_voter': ['c', ['a', 'b']]})), False)
check('The target lacks the leader final log entry', solve(dict(r, **{'log_caught_up': [11, 12]})), False)
check('A delayed handoff names an obsolete term', solve(dict(r, **{'transfer_term': [3, 4]})), False)
check('The handoff arrives after its transfer deadline', solve(dict(r, **{'transfer_deadline': [10, 10]})), False)
check('A transfer races outstanding leader writes', solve(dict(r, **{'writes_drained': 1})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'target_voter': ['c', ['a', 'b']], 'log_caught_up': [11, 12], 'transfer_term': [3, 4], 'transfer_deadline': [10, 10], 'writes_drained': 1}
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 leader transfer operation is admitted even though a transfer races outstanding leader writes.

Return a Boolean admission decision for transfer leadership to a chosen voter. The record r must satisfy all of: r['target_voter'][0] in r['target_voter'][1]; r['log_caught_up'][0] >= r['log_caught_up'][1]; r['transfer_term'][0] == r['transfer_term'][1]; r['transfer_deadline'][0] < r['transfer_deadline'][1]; r['writes_drained'] == 0. 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 (set(r['complete_chunks'][0]) == set(range(r['complete_chunks'][1]))) and (r['checksum_match'][0] == r['checksum_match'][1]) and (bool(r['configuration_present'])) and (r['index_term_pair'][0] == r['index_term_pair'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'newer_applied': [12, 10], 'complete_chunks': [[0, 1, 2], 3], 'checksum_match': ['abc', 'abc'], 'configuration_present': [1, 2, 3], 'index_term_pair': [[12, 3], [12, 3]]}
check('valid operation', solve(r), True)
check('A snapshot rolls back applied state', solve(dict(r, **{'newer_applied': [9, 10]})), False)
check('An incomplete snapshot replaces the live state', solve(dict(r, **{'complete_chunks': [[0, 2], 3]})), False)
check('Corrupt snapshot contents become authoritative', solve(dict(r, **{'checksum_match': ['bad', 'abc']})), False)
check('The snapshot omits its membership configuration', solve(dict(r, **{'configuration_present': []})), False)
check('Snapshot metadata refers to a different log term', solve(dict(r, **{'index_term_pair': [[12, 2], [12, 3]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'newer_applied': [9, 10], 'complete_chunks': [[0, 2], 3], 'checksum_match': ['bad', 'abc'], 'configuration_present': [], 'index_term_pair': [[12, 2], [12, 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-snapshot-install

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

N = 1
observations = []
def solve(r):
    return (r['newer_applied'][0] >= r['newer_applied'][1]) and (r['checksum_match'][0] == r['checksum_match'][1]) and (bool(r['configuration_present'])) and (r['index_term_pair'][0] == r['index_term_pair'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'newer_applied': [12, 10], 'complete_chunks': [[0, 1, 2], 3], 'checksum_match': ['abc', 'abc'], 'configuration_present': [1, 2, 3], 'index_term_pair': [[12, 3], [12, 3]]}
check('valid operation', solve(r), True)
check('A snapshot rolls back applied state', solve(dict(r, **{'newer_applied': [9, 10]})), False)
check('An incomplete snapshot replaces the live state', solve(dict(r, **{'complete_chunks': [[0, 2], 3]})), False)
check('Corrupt snapshot contents become authoritative', solve(dict(r, **{'checksum_match': ['bad', 'abc']})), False)
check('The snapshot omits its membership configuration', solve(dict(r, **{'configuration_present': []})), False)
check('Snapshot metadata refers to a different log term', solve(dict(r, **{'index_term_pair': [[12, 2], [12, 3]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'newer_applied': [9, 10], 'complete_chunks': [[0, 2], 3], 'checksum_match': ['bad', 'abc'], 'configuration_present': [], 'index_term_pair': [[12, 2], [12, 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 snapshot install operation is admitted even though a snapshot rolls back applied state.

Return a Boolean admission decision for install a replicated state snapshot. The record r must satisfy all of: r['newer_applied'][0] >= r['newer_applied'][1]; set(r['complete_chunks'][0]) == set(range(r['complete_chunks'][1])); r['checksum_match'][0] == r['checksum_match'][1]; bool(r['configuration_present']); r['index_term_pair'][0] == r['index_term_pair'][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['newer_applied'][0] >= r['newer_applied'][1]) and (r['checksum_match'][0] == r['checksum_match'][1]) and (bool(r['configuration_present'])) and (r['index_term_pair'][0] == r['index_term_pair'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'newer_applied': [12, 10], 'complete_chunks': [[0, 1, 2], 3], 'checksum_match': ['abc', 'abc'], 'configuration_present': [1, 2, 3], 'index_term_pair': [[12, 3], [12, 3]]}
check('valid operation', solve(r), True)
check('A snapshot rolls back applied state', solve(dict(r, **{'newer_applied': [9, 10]})), False)
check('An incomplete snapshot replaces the live state', solve(dict(r, **{'complete_chunks': [[0, 2], 3]})), False)
check('Corrupt snapshot contents become authoritative', solve(dict(r, **{'checksum_match': ['bad', 'abc']})), False)
check('The snapshot omits its membership configuration', solve(dict(r, **{'configuration_present': []})), False)
check('Snapshot metadata refers to a different log term', solve(dict(r, **{'index_term_pair': [[12, 2], [12, 3]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'newer_applied': [9, 10], 'complete_chunks': [[0, 2], 3], 'checksum_match': ['bad', 'abc'], 'configuration_present': [], 'index_term_pair': [[12, 2], [12, 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-snapshot-install

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

N = 1
observations = []
def solve(r):
    return (r['newer_applied'][0] >= r['newer_applied'][1]) and (set(r['complete_chunks'][0]) == set(range(r['complete_chunks'][1]))) and (bool(r['configuration_present'])) and (r['index_term_pair'][0] == r['index_term_pair'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'newer_applied': [12, 10], 'complete_chunks': [[0, 1, 2], 3], 'checksum_match': ['abc', 'abc'], 'configuration_present': [1, 2, 3], 'index_term_pair': [[12, 3], [12, 3]]}
check('valid operation', solve(r), True)
check('A snapshot rolls back applied state', solve(dict(r, **{'newer_applied': [9, 10]})), False)
check('An incomplete snapshot replaces the live state', solve(dict(r, **{'complete_chunks': [[0, 2], 3]})), False)
check('Corrupt snapshot contents become authoritative', solve(dict(r, **{'checksum_match': ['bad', 'abc']})), False)
check('The snapshot omits its membership configuration', solve(dict(r, **{'configuration_present': []})), False)
check('Snapshot metadata refers to a different log term', solve(dict(r, **{'index_term_pair': [[12, 2], [12, 3]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'newer_applied': [9, 10], 'complete_chunks': [[0, 2], 3], 'checksum_match': ['bad', 'abc'], 'configuration_present': [], 'index_term_pair': [[12, 2], [12, 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 snapshot install operation is admitted even though an incomplete snapshot replaces the live state.

Return a Boolean admission decision for install a replicated state snapshot. The record r must satisfy all of: r['newer_applied'][0] >= r['newer_applied'][1]; set(r['complete_chunks'][0]) == set(range(r['complete_chunks'][1])); r['checksum_match'][0] == r['checksum_match'][1]; bool(r['configuration_present']); r['index_term_pair'][0] == r['index_term_pair'][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['newer_applied'][0] >= r['newer_applied'][1]) and (set(r['complete_chunks'][0]) == set(range(r['complete_chunks'][1]))) and (bool(r['configuration_present'])) and (r['index_term_pair'][0] == r['index_term_pair'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'newer_applied': [12, 10], 'complete_chunks': [[0, 1, 2], 3], 'checksum_match': ['abc', 'abc'], 'configuration_present': [1, 2, 3], 'index_term_pair': [[12, 3], [12, 3]]}
check('valid operation', solve(r), True)
check('A snapshot rolls back applied state', solve(dict(r, **{'newer_applied': [9, 10]})), False)
check('An incomplete snapshot replaces the live state', solve(dict(r, **{'complete_chunks': [[0, 2], 3]})), False)
check('Corrupt snapshot contents become authoritative', solve(dict(r, **{'checksum_match': ['bad', 'abc']})), False)
check('The snapshot omits its membership configuration', solve(dict(r, **{'configuration_present': []})), False)
check('Snapshot metadata refers to a different log term', solve(dict(r, **{'index_term_pair': [[12, 2], [12, 3]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'newer_applied': [9, 10], 'complete_chunks': [[0, 2], 3], 'checksum_match': ['bad', 'abc'], 'configuration_present': [], 'index_term_pair': [[12, 2], [12, 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-snapshot-install

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

N = 1
observations = []
def solve(r):
    return (r['newer_applied'][0] >= r['newer_applied'][1]) and (set(r['complete_chunks'][0]) == set(range(r['complete_chunks'][1]))) and (r['checksum_match'][0] == r['checksum_match'][1]) and (r['index_term_pair'][0] == r['index_term_pair'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'newer_applied': [12, 10], 'complete_chunks': [[0, 1, 2], 3], 'checksum_match': ['abc', 'abc'], 'configuration_present': [1, 2, 3], 'index_term_pair': [[12, 3], [12, 3]]}
check('valid operation', solve(r), True)
check('A snapshot rolls back applied state', solve(dict(r, **{'newer_applied': [9, 10]})), False)
check('An incomplete snapshot replaces the live state', solve(dict(r, **{'complete_chunks': [[0, 2], 3]})), False)
check('Corrupt snapshot contents become authoritative', solve(dict(r, **{'checksum_match': ['bad', 'abc']})), False)
check('The snapshot omits its membership configuration', solve(dict(r, **{'configuration_present': []})), False)
check('Snapshot metadata refers to a different log term', solve(dict(r, **{'index_term_pair': [[12, 2], [12, 3]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'newer_applied': [9, 10], 'complete_chunks': [[0, 2], 3], 'checksum_match': ['bad', 'abc'], 'configuration_present': [], 'index_term_pair': [[12, 2], [12, 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 snapshot install operation is admitted even though corrupt snapshot contents become authoritative.

Return a Boolean admission decision for install a replicated state snapshot. The record r must satisfy all of: r['newer_applied'][0] >= r['newer_applied'][1]; set(r['complete_chunks'][0]) == set(range(r['complete_chunks'][1])); r['checksum_match'][0] == r['checksum_match'][1]; bool(r['configuration_present']); r['index_term_pair'][0] == r['index_term_pair'][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['newer_applied'][0] >= r['newer_applied'][1]) and (set(r['complete_chunks'][0]) == set(range(r['complete_chunks'][1]))) and (r['checksum_match'][0] == r['checksum_match'][1]) and (r['index_term_pair'][0] == r['index_term_pair'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'newer_applied': [12, 10], 'complete_chunks': [[0, 1, 2], 3], 'checksum_match': ['abc', 'abc'], 'configuration_present': [1, 2, 3], 'index_term_pair': [[12, 3], [12, 3]]}
check('valid operation', solve(r), True)
check('A snapshot rolls back applied state', solve(dict(r, **{'newer_applied': [9, 10]})), False)
check('An incomplete snapshot replaces the live state', solve(dict(r, **{'complete_chunks': [[0, 2], 3]})), False)
check('Corrupt snapshot contents become authoritative', solve(dict(r, **{'checksum_match': ['bad', 'abc']})), False)
check('The snapshot omits its membership configuration', solve(dict(r, **{'configuration_present': []})), False)
check('Snapshot metadata refers to a different log term', solve(dict(r, **{'index_term_pair': [[12, 2], [12, 3]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'newer_applied': [9, 10], 'complete_chunks': [[0, 2], 3], 'checksum_match': ['bad', 'abc'], 'configuration_present': [], 'index_term_pair': [[12, 2], [12, 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-snapshot-install

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

N = 1
observations = []
def solve(r):
    return (r['newer_applied'][0] >= r['newer_applied'][1]) and (set(r['complete_chunks'][0]) == set(range(r['complete_chunks'][1]))) and (r['checksum_match'][0] == r['checksum_match'][1]) and (bool(r['configuration_present']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'newer_applied': [12, 10], 'complete_chunks': [[0, 1, 2], 3], 'checksum_match': ['abc', 'abc'], 'configuration_present': [1, 2, 3], 'index_term_pair': [[12, 3], [12, 3]]}
check('valid operation', solve(r), True)
check('A snapshot rolls back applied state', solve(dict(r, **{'newer_applied': [9, 10]})), False)
check('An incomplete snapshot replaces the live state', solve(dict(r, **{'complete_chunks': [[0, 2], 3]})), False)
check('Corrupt snapshot contents become authoritative', solve(dict(r, **{'checksum_match': ['bad', 'abc']})), False)
check('The snapshot omits its membership configuration', solve(dict(r, **{'configuration_present': []})), False)
check('Snapshot metadata refers to a different log term', solve(dict(r, **{'index_term_pair': [[12, 2], [12, 3]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'newer_applied': [9, 10], 'complete_chunks': [[0, 2], 3], 'checksum_match': ['bad', 'abc'], 'configuration_present': [], 'index_term_pair': [[12, 2], [12, 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 snapshot install operation is admitted even though the snapshot omits its membership configuration.

Return a Boolean admission decision for install a replicated state snapshot. The record r must satisfy all of: r['newer_applied'][0] >= r['newer_applied'][1]; set(r['complete_chunks'][0]) == set(range(r['complete_chunks'][1])); r['checksum_match'][0] == r['checksum_match'][1]; bool(r['configuration_present']); r['index_term_pair'][0] == r['index_term_pair'][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['newer_applied'][0] >= r['newer_applied'][1]) and (set(r['complete_chunks'][0]) == set(range(r['complete_chunks'][1]))) and (r['checksum_match'][0] == r['checksum_match'][1]) and (bool(r['configuration_present']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'newer_applied': [12, 10], 'complete_chunks': [[0, 1, 2], 3], 'checksum_match': ['abc', 'abc'], 'configuration_present': [1, 2, 3], 'index_term_pair': [[12, 3], [12, 3]]}
check('valid operation', solve(r), True)
check('A snapshot rolls back applied state', solve(dict(r, **{'newer_applied': [9, 10]})), False)
check('An incomplete snapshot replaces the live state', solve(dict(r, **{'complete_chunks': [[0, 2], 3]})), False)
check('Corrupt snapshot contents become authoritative', solve(dict(r, **{'checksum_match': ['bad', 'abc']})), False)
check('The snapshot omits its membership configuration', solve(dict(r, **{'configuration_present': []})), False)
check('Snapshot metadata refers to a different log term', solve(dict(r, **{'index_term_pair': [[12, 2], [12, 3]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'newer_applied': [9, 10], 'complete_chunks': [[0, 2], 3], 'checksum_match': ['bad', 'abc'], 'configuration_present': [], 'index_term_pair': [[12, 2], [12, 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-snapshot-install

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

N = 1
observations = []
def solve(r):
    return (set(r['complete_chunks'][0]) == set(range(r['complete_chunks'][1]))) and (r['checksum_match'][0] == r['checksum_match'][1]) and (bool(r['configuration_present'])) and (r['index_term_pair'][0] == r['index_term_pair'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'newer_applied': [12, 10], 'complete_chunks': [[0, 1, 2], 3], 'checksum_match': ['abc', 'abc'], 'configuration_present': [1, 2, 3], 'index_term_pair': [[12, 3], [12, 3]]}
check('valid operation', solve(r), True)
check('A snapshot rolls back applied state', solve(dict(r, **{'newer_applied': [9, 10]})), False)
check('An incomplete snapshot replaces the live state', solve(dict(r, **{'complete_chunks': [[0, 2], 3]})), False)
check('Corrupt snapshot contents become authoritative', solve(dict(r, **{'checksum_match': ['bad', 'abc']})), False)
check('The snapshot omits its membership configuration', solve(dict(r, **{'configuration_present': []})), False)
check('Snapshot metadata refers to a different log term', solve(dict(r, **{'index_term_pair': [[12, 2], [12, 3]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'newer_applied': [9, 10], 'complete_chunks': [[0, 2], 3], 'checksum_match': ['bad', 'abc'], 'configuration_present': [], 'index_term_pair': [[12, 2], [12, 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 snapshot install operation is admitted even though snapshot metadata refers to a different log term.

Return a Boolean admission decision for install a replicated state snapshot. The record r must satisfy all of: r['newer_applied'][0] >= r['newer_applied'][1]; set(r['complete_chunks'][0]) == set(range(r['complete_chunks'][1])); r['checksum_match'][0] == r['checksum_match'][1]; bool(r['configuration_present']); r['index_term_pair'][0] == r['index_term_pair'][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['prefix_exists'][0] <= r['prefix_exists'][1]) and (r['prefix_term'][0] == r['prefix_term'][1]) and (r['commit_protected'][0] > r['commit_protected'][1]) and (r['contiguous_indexes'][1] == list(range(r['contiguous_indexes'][0]+1, r['contiguous_indexes'][0]+1+len(r['contiguous_indexes'][1]))))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'leader_term': [5, 5], 'prefix_exists': [8, 8], 'prefix_term': [4, 4], 'commit_protected': [9, 8], 'contiguous_indexes': [8, [9, 10, 11]]}
check('valid operation', solve(r), True)
check('An old leader appends entries after a term change', solve(dict(r, **{'leader_term': [4, 5]})), False)
check('An append references a missing predecessor index', solve(dict(r, **{'prefix_exists': [9, 8]})), False)
check('An append attaches to a conflicting predecessor term', solve(dict(r, **{'prefix_term': [3, 4]})), False)
check('Conflict truncation removes a committed entry', solve(dict(r, **{'commit_protected': [8, 8]})), False)
check('Incoming log entries contain an index gap', solve(dict(r, **{'contiguous_indexes': [8, [9, 11]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'leader_term': [4, 5], 'prefix_exists': [9, 8], 'prefix_term': [3, 4], 'commit_protected': [8, 8], 'contiguous_indexes': [8, [9, 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-append-log

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

N = 1
observations = []
def solve(r):
    return (r['leader_term'][0] >= r['leader_term'][1]) and (r['prefix_term'][0] == r['prefix_term'][1]) and (r['commit_protected'][0] > r['commit_protected'][1]) and (r['contiguous_indexes'][1] == list(range(r['contiguous_indexes'][0]+1, r['contiguous_indexes'][0]+1+len(r['contiguous_indexes'][1]))))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'leader_term': [5, 5], 'prefix_exists': [8, 8], 'prefix_term': [4, 4], 'commit_protected': [9, 8], 'contiguous_indexes': [8, [9, 10, 11]]}
check('valid operation', solve(r), True)
check('An old leader appends entries after a term change', solve(dict(r, **{'leader_term': [4, 5]})), False)
check('An append references a missing predecessor index', solve(dict(r, **{'prefix_exists': [9, 8]})), False)
check('An append attaches to a conflicting predecessor term', solve(dict(r, **{'prefix_term': [3, 4]})), False)
check('Conflict truncation removes a committed entry', solve(dict(r, **{'commit_protected': [8, 8]})), False)
check('Incoming log entries contain an index gap', solve(dict(r, **{'contiguous_indexes': [8, [9, 11]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'leader_term': [4, 5], 'prefix_exists': [9, 8], 'prefix_term': [3, 4], 'commit_protected': [8, 8], 'contiguous_indexes': [8, [9, 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 append log operation is admitted even though an old leader appends entries after a term change.

Return a Boolean admission decision for admit a leader log append. The record r must satisfy all of: r['leader_term'][0] >= r['leader_term'][1]; r['prefix_exists'][0] <= r['prefix_exists'][1]; r['prefix_term'][0] == r['prefix_term'][1]; r['commit_protected'][0] > r['commit_protected'][1]; r['contiguous_indexes'][1] == list(range(r['contiguous_indexes'][0]+1, r['contiguous_indexes'][0]+1+len(r['contiguous_indexes'][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['leader_term'][0] >= r['leader_term'][1]) and (r['prefix_term'][0] == r['prefix_term'][1]) and (r['commit_protected'][0] > r['commit_protected'][1]) and (r['contiguous_indexes'][1] == list(range(r['contiguous_indexes'][0]+1, r['contiguous_indexes'][0]+1+len(r['contiguous_indexes'][1]))))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'leader_term': [5, 5], 'prefix_exists': [8, 8], 'prefix_term': [4, 4], 'commit_protected': [9, 8], 'contiguous_indexes': [8, [9, 10, 11]]}
check('valid operation', solve(r), True)
check('An old leader appends entries after a term change', solve(dict(r, **{'leader_term': [4, 5]})), False)
check('An append references a missing predecessor index', solve(dict(r, **{'prefix_exists': [9, 8]})), False)
check('An append attaches to a conflicting predecessor term', solve(dict(r, **{'prefix_term': [3, 4]})), False)
check('Conflict truncation removes a committed entry', solve(dict(r, **{'commit_protected': [8, 8]})), False)
check('Incoming log entries contain an index gap', solve(dict(r, **{'contiguous_indexes': [8, [9, 11]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'leader_term': [4, 5], 'prefix_exists': [9, 8], 'prefix_term': [3, 4], 'commit_protected': [8, 8], 'contiguous_indexes': [8, [9, 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-append-log

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

N = 1
observations = []
def solve(r):
    return (r['leader_term'][0] >= r['leader_term'][1]) and (r['prefix_exists'][0] <= r['prefix_exists'][1]) and (r['commit_protected'][0] > r['commit_protected'][1]) and (r['contiguous_indexes'][1] == list(range(r['contiguous_indexes'][0]+1, r['contiguous_indexes'][0]+1+len(r['contiguous_indexes'][1]))))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'leader_term': [5, 5], 'prefix_exists': [8, 8], 'prefix_term': [4, 4], 'commit_protected': [9, 8], 'contiguous_indexes': [8, [9, 10, 11]]}
check('valid operation', solve(r), True)
check('An old leader appends entries after a term change', solve(dict(r, **{'leader_term': [4, 5]})), False)
check('An append references a missing predecessor index', solve(dict(r, **{'prefix_exists': [9, 8]})), False)
check('An append attaches to a conflicting predecessor term', solve(dict(r, **{'prefix_term': [3, 4]})), False)
check('Conflict truncation removes a committed entry', solve(dict(r, **{'commit_protected': [8, 8]})), False)
check('Incoming log entries contain an index gap', solve(dict(r, **{'contiguous_indexes': [8, [9, 11]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'leader_term': [4, 5], 'prefix_exists': [9, 8], 'prefix_term': [3, 4], 'commit_protected': [8, 8], 'contiguous_indexes': [8, [9, 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 append log operation is admitted even though an append references a missing predecessor index.

Return a Boolean admission decision for admit a leader log append. The record r must satisfy all of: r['leader_term'][0] >= r['leader_term'][1]; r['prefix_exists'][0] <= r['prefix_exists'][1]; r['prefix_term'][0] == r['prefix_term'][1]; r['commit_protected'][0] > r['commit_protected'][1]; r['contiguous_indexes'][1] == list(range(r['contiguous_indexes'][0]+1, r['contiguous_indexes'][0]+1+len(r['contiguous_indexes'][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['leader_term'][0] >= r['leader_term'][1]) and (r['prefix_exists'][0] <= r['prefix_exists'][1]) and (r['commit_protected'][0] > r['commit_protected'][1]) and (r['contiguous_indexes'][1] == list(range(r['contiguous_indexes'][0]+1, r['contiguous_indexes'][0]+1+len(r['contiguous_indexes'][1]))))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'leader_term': [5, 5], 'prefix_exists': [8, 8], 'prefix_term': [4, 4], 'commit_protected': [9, 8], 'contiguous_indexes': [8, [9, 10, 11]]}
check('valid operation', solve(r), True)
check('An old leader appends entries after a term change', solve(dict(r, **{'leader_term': [4, 5]})), False)
check('An append references a missing predecessor index', solve(dict(r, **{'prefix_exists': [9, 8]})), False)
check('An append attaches to a conflicting predecessor term', solve(dict(r, **{'prefix_term': [3, 4]})), False)
check('Conflict truncation removes a committed entry', solve(dict(r, **{'commit_protected': [8, 8]})), False)
check('Incoming log entries contain an index gap', solve(dict(r, **{'contiguous_indexes': [8, [9, 11]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'leader_term': [4, 5], 'prefix_exists': [9, 8], 'prefix_term': [3, 4], 'commit_protected': [8, 8], 'contiguous_indexes': [8, [9, 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-append-log

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

N = 1
observations = []
def solve(r):
    return (r['leader_term'][0] >= r['leader_term'][1]) and (r['prefix_exists'][0] <= r['prefix_exists'][1]) and (r['prefix_term'][0] == r['prefix_term'][1]) and (r['contiguous_indexes'][1] == list(range(r['contiguous_indexes'][0]+1, r['contiguous_indexes'][0]+1+len(r['contiguous_indexes'][1]))))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'leader_term': [5, 5], 'prefix_exists': [8, 8], 'prefix_term': [4, 4], 'commit_protected': [9, 8], 'contiguous_indexes': [8, [9, 10, 11]]}
check('valid operation', solve(r), True)
check('An old leader appends entries after a term change', solve(dict(r, **{'leader_term': [4, 5]})), False)
check('An append references a missing predecessor index', solve(dict(r, **{'prefix_exists': [9, 8]})), False)
check('An append attaches to a conflicting predecessor term', solve(dict(r, **{'prefix_term': [3, 4]})), False)
check('Conflict truncation removes a committed entry', solve(dict(r, **{'commit_protected': [8, 8]})), False)
check('Incoming log entries contain an index gap', solve(dict(r, **{'contiguous_indexes': [8, [9, 11]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'leader_term': [4, 5], 'prefix_exists': [9, 8], 'prefix_term': [3, 4], 'commit_protected': [8, 8], 'contiguous_indexes': [8, [9, 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 append log operation is admitted even though an append attaches to a conflicting predecessor term.

Return a Boolean admission decision for admit a leader log append. The record r must satisfy all of: r['leader_term'][0] >= r['leader_term'][1]; r['prefix_exists'][0] <= r['prefix_exists'][1]; r['prefix_term'][0] == r['prefix_term'][1]; r['commit_protected'][0] > r['commit_protected'][1]; r['contiguous_indexes'][1] == list(range(r['contiguous_indexes'][0]+1, r['contiguous_indexes'][0]+1+len(r['contiguous_indexes'][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['leader_term'][0] >= r['leader_term'][1]) and (r['prefix_exists'][0] <= r['prefix_exists'][1]) and (r['prefix_term'][0] == r['prefix_term'][1]) and (r['contiguous_indexes'][1] == list(range(r['contiguous_indexes'][0]+1, r['contiguous_indexes'][0]+1+len(r['contiguous_indexes'][1]))))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'leader_term': [5, 5], 'prefix_exists': [8, 8], 'prefix_term': [4, 4], 'commit_protected': [9, 8], 'contiguous_indexes': [8, [9, 10, 11]]}
check('valid operation', solve(r), True)
check('An old leader appends entries after a term change', solve(dict(r, **{'leader_term': [4, 5]})), False)
check('An append references a missing predecessor index', solve(dict(r, **{'prefix_exists': [9, 8]})), False)
check('An append attaches to a conflicting predecessor term', solve(dict(r, **{'prefix_term': [3, 4]})), False)
check('Conflict truncation removes a committed entry', solve(dict(r, **{'commit_protected': [8, 8]})), False)
check('Incoming log entries contain an index gap', solve(dict(r, **{'contiguous_indexes': [8, [9, 11]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'leader_term': [4, 5], 'prefix_exists': [9, 8], 'prefix_term': [3, 4], 'commit_protected': [8, 8], 'contiguous_indexes': [8, [9, 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-append-log

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

N = 1
observations = []
def solve(r):
    return (r['leader_term'][0] >= r['leader_term'][1]) and (r['prefix_exists'][0] <= r['prefix_exists'][1]) and (r['prefix_term'][0] == r['prefix_term'][1]) and (r['commit_protected'][0] > r['commit_protected'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'leader_term': [5, 5], 'prefix_exists': [8, 8], 'prefix_term': [4, 4], 'commit_protected': [9, 8], 'contiguous_indexes': [8, [9, 10, 11]]}
check('valid operation', solve(r), True)
check('An old leader appends entries after a term change', solve(dict(r, **{'leader_term': [4, 5]})), False)
check('An append references a missing predecessor index', solve(dict(r, **{'prefix_exists': [9, 8]})), False)
check('An append attaches to a conflicting predecessor term', solve(dict(r, **{'prefix_term': [3, 4]})), False)
check('Conflict truncation removes a committed entry', solve(dict(r, **{'commit_protected': [8, 8]})), False)
check('Incoming log entries contain an index gap', solve(dict(r, **{'contiguous_indexes': [8, [9, 11]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'leader_term': [4, 5], 'prefix_exists': [9, 8], 'prefix_term': [3, 4], 'commit_protected': [8, 8], 'contiguous_indexes': [8, [9, 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 append log operation is admitted even though conflict truncation removes a committed entry.

Return a Boolean admission decision for admit a leader log append. The record r must satisfy all of: r['leader_term'][0] >= r['leader_term'][1]; r['prefix_exists'][0] <= r['prefix_exists'][1]; r['prefix_term'][0] == r['prefix_term'][1]; r['commit_protected'][0] > r['commit_protected'][1]; r['contiguous_indexes'][1] == list(range(r['contiguous_indexes'][0]+1, r['contiguous_indexes'][0]+1+len(r['contiguous_indexes'][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['leader_term'][0] >= r['leader_term'][1]) and (r['prefix_exists'][0] <= r['prefix_exists'][1]) and (r['prefix_term'][0] == r['prefix_term'][1]) and (r['commit_protected'][0] > r['commit_protected'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'leader_term': [5, 5], 'prefix_exists': [8, 8], 'prefix_term': [4, 4], 'commit_protected': [9, 8], 'contiguous_indexes': [8, [9, 10, 11]]}
check('valid operation', solve(r), True)
check('An old leader appends entries after a term change', solve(dict(r, **{'leader_term': [4, 5]})), False)
check('An append references a missing predecessor index', solve(dict(r, **{'prefix_exists': [9, 8]})), False)
check('An append attaches to a conflicting predecessor term', solve(dict(r, **{'prefix_term': [3, 4]})), False)
check('Conflict truncation removes a committed entry', solve(dict(r, **{'commit_protected': [8, 8]})), False)
check('Incoming log entries contain an index gap', solve(dict(r, **{'contiguous_indexes': [8, [9, 11]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'leader_term': [4, 5], 'prefix_exists': [9, 8], 'prefix_term': [3, 4], 'commit_protected': [8, 8], 'contiguous_indexes': [8, [9, 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-append-log

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

N = 1
observations = []
def solve(r):
    return (r['prefix_exists'][0] <= r['prefix_exists'][1]) and (r['prefix_term'][0] == r['prefix_term'][1]) and (r['commit_protected'][0] > r['commit_protected'][1]) and (r['contiguous_indexes'][1] == list(range(r['contiguous_indexes'][0]+1, r['contiguous_indexes'][0]+1+len(r['contiguous_indexes'][1]))))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'leader_term': [5, 5], 'prefix_exists': [8, 8], 'prefix_term': [4, 4], 'commit_protected': [9, 8], 'contiguous_indexes': [8, [9, 10, 11]]}
check('valid operation', solve(r), True)
check('An old leader appends entries after a term change', solve(dict(r, **{'leader_term': [4, 5]})), False)
check('An append references a missing predecessor index', solve(dict(r, **{'prefix_exists': [9, 8]})), False)
check('An append attaches to a conflicting predecessor term', solve(dict(r, **{'prefix_term': [3, 4]})), False)
check('Conflict truncation removes a committed entry', solve(dict(r, **{'commit_protected': [8, 8]})), False)
check('Incoming log entries contain an index gap', solve(dict(r, **{'contiguous_indexes': [8, [9, 11]]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'leader_term': [4, 5], 'prefix_exists': [9, 8], 'prefix_term': [3, 4], 'commit_protected': [8, 8], 'contiguous_indexes': [8, [9, 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 append log operation is admitted even though incoming log entries contain an index gap.

Return a Boolean admission decision for admit a leader log append. The record r must satisfy all of: r['leader_term'][0] >= r['leader_term'][1]; r['prefix_exists'][0] <= r['prefix_exists'][1]; r['prefix_term'][0] == r['prefix_term'][1]; r['commit_protected'][0] > r['commit_protected'][1]; r['contiguous_indexes'][1] == list(range(r['contiguous_indexes'][0]+1, r['contiguous_indexes'][0]+1+len(r['contiguous_indexes'][1]))). Extra tracing fields are ignored; validation does not mutate the record.

split: open-access