FA-881 / Queues / Open access
Commit a consume-transform-publish transaction: Input progress commits before the transformed output · case 01
The transactional consume operation is admitted even though input progress commits before the transformed output.
ROOT CAUSE
The admission path omits the output committed invariant while validating the other operation preconditions.
VERIFIED REPAIR
Require r['output_committed'] is True together with every other stated precondition before accepting the operation.
Unsuccessful approach: Adding the output committed check repairs the reported defect, but replacing the adjacent offset exact check loses that independent invariant.
Case contract
Return a Boolean admission decision for commit a consume-transform-publish transaction. The record r must satisfy all of: r['input_owned'][0] == r['input_owned'][1]; r['output_committed'] is True; r['offset_exact'][0] == r['offset_exact'][1]+1; r['transaction_open'] == 'open'; r['producer_fenced'][0] == r['producer_fenced'][1]. Extra tracing fields are ignored; validation does not mutate the record.
Why this case matters
A deterministic local contract for queues. Each negative fixture violates exactly one invariant. No transport timing, persistence, cryptographic verification, or full protocol implementation is claimed.
1 / The failure
Exit 1"""Failure Map reference implementation. Python standard library only."""
import json
N = 1
observations = []
def solve(r):
return (r['input_owned'][0] == r['input_owned'][1]) and (r['offset_exact'][0] == r['offset_exact'][1]+1) and (r['transaction_open'] == 'open') and (r['producer_fenced'][0] == r['producer_fenced'][1])
def check(label, actual, expected):
observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'input_owned': [8, 8], 'output_committed': True, 'offset_exact': [11, 10], 'transaction_open': 'open', 'producer_fenced': [9, 9]}
check('valid operation', solve(r), True)
check('Commit acknowledges an input partition no longer owned', solve(dict(r, **{'input_owned': [7, 8]})), False)
check('Input progress commits before the transformed output', solve(dict(r, **{'output_committed': False})), False)
check('Commit skips an unprocessed input offset', solve(dict(r, **{'offset_exact': [12, 10]})), False)
check('A commit revives an aborted consume transaction', solve(dict(r, **{'transaction_open': 'aborted'})), False)
check('A previous producer instance commits an old transaction', solve(dict(r, **{'producer_fenced': [8, 9]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'input_owned': [7, 8], 'output_committed': False, 'offset_exact': [12, 10], 'transaction_open': 'aborted', 'producer_fenced': [8, 9]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)
| Boundary fixture | Actual | Expected | Outcome |
|---|---|---|---|
| valid operation | True | True | Passed |
| Commit acknowledges an input partition no longer owned | False | False | Passed |
| Input progress commits before the transformed output | True | False | Failed |
| Commit skips an unprocessed input offset | False | False | Passed |
| A commit revives an aborted consume transaction | False | False | Passed |
| A previous producer instance commits an old transaction | False | False | Passed |
| unrelated tracing metadata | True | True | Passed |
| repeat validation is pure | True | True | Passed |
| two independent violations in variant | False | False | Passed |
SHA-256 / c0f0bb612232cfbaea7cc1527e4ad5866c88eba5e69f2e78ed6c0121312926cd
2 / The unsuccessful fix
Exit 1"""Failure Map reference implementation. Python standard library only."""
import json
N = 1
observations = []
def solve(r):
return (r['input_owned'][0] == r['input_owned'][1]) and (r['output_committed'] is True) and (r['transaction_open'] == 'open') and (r['producer_fenced'][0] == r['producer_fenced'][1])
def check(label, actual, expected):
observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'input_owned': [8, 8], 'output_committed': True, 'offset_exact': [11, 10], 'transaction_open': 'open', 'producer_fenced': [9, 9]}
check('valid operation', solve(r), True)
check('Commit acknowledges an input partition no longer owned', solve(dict(r, **{'input_owned': [7, 8]})), False)
check('Input progress commits before the transformed output', solve(dict(r, **{'output_committed': False})), False)
check('Commit skips an unprocessed input offset', solve(dict(r, **{'offset_exact': [12, 10]})), False)
check('A commit revives an aborted consume transaction', solve(dict(r, **{'transaction_open': 'aborted'})), False)
check('A previous producer instance commits an old transaction', solve(dict(r, **{'producer_fenced': [8, 9]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'input_owned': [7, 8], 'output_committed': False, 'offset_exact': [12, 10], 'transaction_open': 'aborted', 'producer_fenced': [8, 9]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)
| Boundary fixture | Actual | Expected | Outcome |
|---|---|---|---|
| valid operation | True | True | Passed |
| Commit acknowledges an input partition no longer owned | False | False | Passed |
| Input progress commits before the transformed output | False | False | Passed |
| Commit skips an unprocessed input offset | True | False | Failed |
| A commit revives an aborted consume transaction | False | False | Passed |
| A previous producer instance commits an old transaction | False | False | Passed |
| unrelated tracing metadata | True | True | Passed |
| repeat validation is pure | True | True | Passed |
| two independent violations in variant | False | False | Passed |
SHA-256 / b00dcd097535bcae98c2e5bdd4fcba124073f4af364d561b68ae920ae1409632
3 / The verified repair
Exit 0"""Failure Map reference implementation. Python standard library only."""
import json
N = 1
observations = []
def solve(r):
return (r['input_owned'][0] == r['input_owned'][1]) and (r['output_committed'] is True) and (r['offset_exact'][0] == r['offset_exact'][1]+1) and (r['transaction_open'] == 'open') and (r['producer_fenced'][0] == r['producer_fenced'][1])
def check(label, actual, expected):
observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'input_owned': [8, 8], 'output_committed': True, 'offset_exact': [11, 10], 'transaction_open': 'open', 'producer_fenced': [9, 9]}
check('valid operation', solve(r), True)
check('Commit acknowledges an input partition no longer owned', solve(dict(r, **{'input_owned': [7, 8]})), False)
check('Input progress commits before the transformed output', solve(dict(r, **{'output_committed': False})), False)
check('Commit skips an unprocessed input offset', solve(dict(r, **{'offset_exact': [12, 10]})), False)
check('A commit revives an aborted consume transaction', solve(dict(r, **{'transaction_open': 'aborted'})), False)
check('A previous producer instance commits an old transaction', solve(dict(r, **{'producer_fenced': [8, 9]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'input_owned': [7, 8], 'output_committed': False, 'offset_exact': [12, 10], 'transaction_open': 'aborted', 'producer_fenced': [8, 9]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)
| Boundary fixture | Actual | Expected | Outcome |
|---|---|---|---|
| valid operation | True | True | Passed |
| Commit acknowledges an input partition no longer owned | False | False | Passed |
| Input progress commits before the transformed output | False | False | Passed |
| Commit skips an unprocessed input offset | False | False | Passed |
| A commit revives an aborted consume transaction | False | False | Passed |
| A previous producer instance commits an old transaction | False | False | Passed |
| unrelated tracing metadata | True | True | Passed |
| repeat validation is pure | True | True | Passed |
| two independent violations in variant | False | False | Passed |
SHA-256 / 80e85882f5fbe52ad3183d96099fea52444649192c9fd9bdb3f60b1664a6dce6
Verification & scope
This reproducer isolates one failure mechanism. Results cover the supplied fixtures. Variants within a family share a test contract and should remain grouped when constructing evaluation splits. Related mechanisms with a shared evaluation_group must also remain together; these controlled models are not independent production incidents.
Observations recorded using Python 3.12.14 at 2026-09-29T14:36:56.573320+00:00.
Case digest / 8a95544c8fbf2b0594fd27121e4fe33634c4242edb6b61dd6709417bdbbbbe30