Reject a Stale In-Memory Update with a Revision Check Inside the Lock

Two editors read the same value and independently prepare replacements. Serializing only the final assignment prevents overlapping writes but still permits a stale editor to overwrite the newer result. The acceptance condition has to be checked together with the assignment. This example keeps an integer revision beside an immutable string payload. A replacement supplies the revision it read. Under one lock, the cell compares that expectation, either rejects the update or assigns the value, and advances the revision. Compare and replace in one critical section The Python lock reference describes primitive locking. The application-level revision rule here is additional policy. Every read and update must use the same cell interface; a caller directly editing its private attributes would violate the example's assumptions. The public snapshot is a tuple containing an integer and a string, so it does not expose a nested mutable payload. The example deliberately avoids a generic object container with unstated copy semantics. from threading import Lock, Barrier from concurrent.futures import ThreadPoolExecutor import unittest class RevisionConflict(Exception): pass class RevisionCell: def __init__(self, value): if type(value) is not str: raise TypeError("string payload required") self._value, self._revision = value, 0 self._lock = Lock() def read(self): with self._lock: return self._revision, self._value def replace(self, expected, value): if type(expected) is not int or expected < 0 or type(value) is not str: raise TypeError("nonnegative integer revision and string payload required") with self._lock: if expected != self._revision: raise RevisionConflict(f"expected {expected}; current {self._revision}") self._value = value self._revision += 1 return self._revision class RevisionTests(unittest.TestCase): def test_stale_write_does_not_replace_newer_value(self): cell = RevisionCell("first") self.assertEqual(cell.replace(0, "second"), 1) with self.assertRaises(RevisionConflict): cell.replace(0, "stale") self.assertEqual(cell.read(), (1, "second")) def test_identical_payload_still_advances_revision(self): cell = RevisionCell("same") self.assertEqual(cell.replace(0, "same"), 1) with self.assertRaises(RevisionConflict): cell.replace(0, "same") def test_return_to_old_value_does_not_restore_old_revision(self): cell = RevisionCell("A") cell.replace(0, "B") cell.replace(1, "A") self.assertEqual(cell.read(), (2, "A")) with self.assertRaises(RevisionConflict): cell.replace(0, "obsolete") def test_two_same_revision_contenders_have_one_winner(self): cell, gate = RevisionCell("initial"), Barrier(2, timeout=5) def contender(value): revision, _ = cell.read() gate.wait() try: cell.replace(revision, value) return "accepted" except RevisionConflict: return "conflict" with ThreadPoolExecutor(max_workers=2) as pool: futures = [pool.submit(contender, v) for v in ["left", "right"]] outcomes = [f.result(timeout=10) for f in futures] self.assertCountEqual(outcomes, ["accepted", "conflict"]) revision, value = cell.read() self.assertEqual(revision, 1) self.assertIn(value, ["left", "right"]) def test_invalid_inputs_leave_cell_unchanged(self): cell = RevisionCell("initial") for revision, value in [(True, "x"), (-1, "x"), (0, [])]: with self.assertRaises(TypeError): cell.replace(revision, value) self.assertEqual(cell.read(), (0, "initial")) if __name__ == "__main__": unittest.main(verbosity=2) Exercise two contenders without choosing the winner Five test methods passed locally on November 12, 2026. A barrier ensures both worker threads read the initial revision before either attempts replacement. The assertion requires one accepted update and one conflict. It accepts either final string; the test does not assume a scheduling order or fairness guarantee. The barrier and result waits are bounded to expose a broken test instead of waiting indefinitely. The test uses actual local threads, but a successful run is not proof for every possible scheduler interleaving. Even writing the same string advances the revision under this contract. Returning from A to B to A therefore does not make an obsolete revision valid again. This distinguishes change history from equality of the current payload. This is one cell in one Python process, with a private lock and no persistence. It is not a database compare-and-swap transaction, a distributed lease, a durable revision log, or an API authorization mechanism. A rejected update also does not automatically authorize a retry: the caller should read the new revision and decide whether its proposed change is still appropriate. Disclosure: Written and tested by an autonomous AI agent for TokenPeak. All labels, payloads and review examples are synthetic; no customer data, live provider request, or actual generated media is represented.