------------------------------ MODULE Remora ------------------------------ EXTENDS Naturals, FiniteSets, TLC \* JobProtocol: distributed job coordination for Remora (see docs/spec/architecture.md). \* \* DurableStore variables are sole authority for phase, cancellation, attempt \* allocation, fencing tokens, and committed results. CoordinationStore variables \* are reconstructible: queues, reservations, leases, cancel notices, live progress. \* Attempts remain observable after losing authority so delayed outcomes and zombie \* target writes are checked. An attempt staging identity abstracts its immutable \* member set and result manifest; only the durable AcceptSuccess compare-and-swap \* selects that manifest descriptor as the CommittedResult. \* \* Abstractions and bounds: \* - Jobs in Init are already durably accepted (submission outbox precedes queueing). \* - Reconciler projection (Enqueue, PublishCancelNotice, CleanTerminalCoordination, \* ClearStaleLiveProgress) is ambient: CoordinationStore access independent of \* per-worker links. Worker-facing coord ops still require CanReach(w, CoordSvc). \* - Live progress is coordination-only; durable steps never clear it. Stale hints \* linger until ClearStaleLiveProgress, ActivateAttempt overwrite, terminal clean, \* or full coord wipe — not co-stepped with Postgres CAS. \* - MaxCoordLosses bounds full CoordinationStore wipes; MaxPartitions bounds \* DropLink events (re-partition of LinkUp or LinkRecovered is allowed within budget). \* - Worker ids are incarnations: WorkerUnused -> WorkerUp is scale-up/restart; \* WorkerDown is permanent for that id. CrashWorker will not crash the last \* remaining Up/Unused incarnation so safety models stay free of total capacity loss. \* - CHECK_DEADLOCK uses Done at full SystemQuiescent (TLC sink only; not a \* product SLO — residual staging/coord is best-effort GC, not safety). CONSTANTS Jobs, Workers, InitialWorkers, NoJob, NoWorker, MaxAttempts, MaxProgress, LeaseLimit, LocalLimit, GlobalLimit, FaultsEnabled, FaultServices, MaxCoordLosses, MaxPartitions ASSUME /\ Jobs # {} /\ Workers # {} /\ InitialWorkers \subseteq Workers /\ InitialWorkers # {} /\ MaxAttempts \in Nat \ {0} /\ MaxProgress \in Nat \ {0} /\ LeaseLimit \in Nat \ {0} /\ LocalLimit \in Nat \ {0} /\ GlobalLimit \in Nat \ {0} /\ FaultsEnabled \in BOOLEAN /\ FaultServices \subseteq {"Coordination", "Durable", "Target"} /\ MaxCoordLosses \in Nat /\ MaxPartitions \in Nat /\ NoJob \notin Jobs /\ NoWorker \notin Workers Queued == "Queued" Running == "Running" RetryWait == "RetryWait" Cancelling == "Cancelling" Succeeded == "Succeeded" Failed == "Failed" Cancelled == "Cancelled" Phases == {Queued, Running, RetryWait, Cancelling, Succeeded, Failed, Cancelled} TerminalPhases == {Succeeded, Failed, Cancelled} WorkerUnused == "WorkerUnused" WorkerUp == "WorkerUp" WorkerDown == "WorkerDown" WorkerStates == {WorkerUnused, WorkerUp, WorkerDown} CoordSvc == "Coordination" DurableSvc == "Durable" TargetSvc == "Target" Services == {CoordSvc, DurableSvc, TargetSvc} LinkUp == "LinkUp" LinkDown == "LinkDown" LinkRecovered == "LinkRecovered" LinkStates == {LinkUp, LinkDown, LinkRecovered} AttemptUnused == "AttemptUnused" AttemptAuthorized == "AttemptAuthorized" AttemptExecuting == "AttemptExecuting" AttemptStopped == "AttemptStopped" AttemptCommitted == "AttemptCommitted" AttemptStates == {AttemptUnused, AttemptAuthorized, AttemptExecuting, AttemptStopped, AttemptCommitted} NoLease == "NoLease" LeaseLive == "LeaseLive" LeaseExpired == "LeaseExpired" LeaseStates == {NoLease, LeaseLive, LeaseExpired} StageNone == "StageNone" StagePartial == "StagePartial" StageComplete == "StageComplete" StageCleaned == "StageCleaned" StageStates == {StageNone, StagePartial, StageComplete, StageCleaned} NoOutcome == "NoOutcome" WantsSuccess == "WantsSuccess" WantsRetryableFailure == "WantsRetryableFailure" WantsPermanentFailure == "WantsPermanentFailure" FailureOutcomes == {WantsRetryableFailure, WantsPermanentFailure} OutcomeRequests == {NoOutcome, WantsSuccess} \cup FailureOutcomes NoAttempt == [job |-> NoJob, token |-> 0] \* Product: opaque attempt id + separate fence; model fuses them as <>. AttemptIds == {[job |-> j, token |-> token] : j \in Jobs, token \in 1..MaxAttempts} AttemptId(j, token) == [job |-> j, token |-> token] JobOf(a) == a.job TokenOf(a) == a.token VARIABLES \* DurableStore authority. phase, cancelRequested, currentToken, authority, dispatchNeeded, committedAttempt, checkpoint, checkpointValid, \* CoordinationStore (reconstructible). queued, reservation, leaseState, leaseAge, cancelNotice, liveProgress, liveProgressAttempt, coordLossCount, \* Worker incarnations, per-service links, partition budget. workerState, links, partitionCount, \* Attempts, staging, delayed outcome requests. attemptState, attemptWorker, attemptProgress, initialProgress, resumedFromCheckpoint, stageState, pendingOutcome durableVars == <> coordinationVars == <> workerVars == <> attemptVars == <> vars == <> Terminal(j) == phase[j] \in TerminalPhases CanReach(w, service) == links[w][service] # LinkDown CurrentAuthority(a) == /\ authority[JobOf(a)] = a /\ currentToken[JobOf(a)] = TokenOf(a) ReservedJobs == {j \in Jobs : reservation[j] # NoWorker} LiveLeases == {a \in AttemptIds : leaseState[a] = LeaseLive} ReservedFor(w) == {j \in Jobs : reservation[j] = w} LeasedFor(w) == {a \in AttemptIds : leaseState[a] = LeaseLive /\ attemptWorker[a] = w} ExecutingFor(w) == {a \in AttemptIds : attemptState[a] = AttemptExecuting /\ attemptWorker[a] = w} CapacityAvailable(w) == /\ Cardinality(ReservedJobs) + Cardinality(LiveLeases) < GlobalLimit /\ Cardinality(ReservedFor(w)) + Cardinality(ExecutingFor(w)) < LocalLimit OtherCapacity(w) == \E w2 \in Workers : /\ w2 # w /\ workerState[w2] \in {WorkerUp, WorkerUnused} CoordinationClean(j) == /\ ~queued[j] /\ reservation[j] = NoWorker /\ ~cancelNotice[j] /\ liveProgress[j] = 0 /\ liveProgressAttempt[j] = NoAttempt \* TLC full quiescence (terminal + coord clean + non-winner staging cleaned). \* Product may retain orphans; do not assert this at runtime. Safety is \* CommittedResult + StagingOwnership, not zero residual objects. SystemQuiescent == /\ \A j \in Jobs : Terminal(j) /\ CoordinationClean(j) /\ \A a \in AttemptIds : /\ leaseState[a] # LeaseLive /\ pendingOutcome[a] = NoOutcome /\ attemptState[a] \notin {AttemptAuthorized, AttemptExecuting} /\ (stageState[a] \in {StagePartial, StageComplete} => committedAttempt[JobOf(a)] = a) Init == /\ phase = [j \in Jobs |-> Queued] /\ cancelRequested = [j \in Jobs |-> FALSE] /\ currentToken = [j \in Jobs |-> 0] /\ authority = [j \in Jobs |-> NoAttempt] /\ dispatchNeeded = [j \in Jobs |-> TRUE] /\ committedAttempt = [j \in Jobs |-> NoAttempt] /\ checkpoint = [j \in Jobs |-> 0] /\ checkpointValid = [j \in Jobs |-> FALSE] /\ queued = [j \in Jobs |-> FALSE] /\ reservation = [j \in Jobs |-> NoWorker] /\ leaseState = [a \in AttemptIds |-> NoLease] /\ leaseAge = [a \in AttemptIds |-> 0] /\ cancelNotice = [j \in Jobs |-> FALSE] /\ liveProgress = [j \in Jobs |-> 0] /\ liveProgressAttempt = [j \in Jobs |-> NoAttempt] /\ coordLossCount = 0 /\ workerState = [w \in Workers |-> IF w \in InitialWorkers THEN WorkerUp ELSE WorkerUnused] /\ links = [w \in Workers |-> [s \in Services |-> LinkUp]] /\ partitionCount = 0 /\ attemptState = [a \in AttemptIds |-> AttemptUnused] /\ attemptWorker = [a \in AttemptIds |-> NoWorker] /\ attemptProgress = [a \in AttemptIds |-> 0] /\ initialProgress = [a \in AttemptIds |-> 0] /\ resumedFromCheckpoint = [a \in AttemptIds |-> FALSE] /\ stageState = [a \in AttemptIds |-> StageNone] /\ pendingOutcome = [a \in AttemptIds |-> NoOutcome] \* Ambient reconciler: projects durable dispatch into CoordinationStore. Enqueue(j) == /\ phase[j] = Queued /\ dispatchNeeded[j] /\ ~queued[j] /\ reservation[j] = NoWorker /\ queued' = [queued EXCEPT ![j] = TRUE] /\ UNCHANGED durableVars /\ UNCHANGED <> /\ UNCHANGED workerVars /\ UNCHANGED attemptVars Reserve(j, w) == /\ queued[j] /\ reservation[j] = NoWorker /\ workerState[w] = WorkerUp /\ CanReach(w, CoordSvc) /\ CapacityAvailable(w) /\ queued' = [queued EXCEPT ![j] = FALSE] /\ reservation' = [reservation EXCEPT ![j] = w] /\ UNCHANGED durableVars /\ UNCHANGED <> /\ UNCHANGED workerVars /\ UNCHANGED attemptVars \* Durable CAS: next attempt and fencing token; consumes retry budget. AuthorizeAttempt(j, w) == LET a == AttemptId(j, currentToken[j] + 1) IN /\ phase[j] = Queued /\ dispatchNeeded[j] /\ reservation[j] = w /\ workerState[w] = WorkerUp /\ CanReach(w, DurableSvc) /\ currentToken[j] < MaxAttempts /\ attemptState[a] = AttemptUnused /\ phase' = [phase EXCEPT ![j] = Running] /\ currentToken' = [currentToken EXCEPT ![j] = @ + 1] /\ authority' = [authority EXCEPT ![j] = a] /\ dispatchNeeded' = [dispatchNeeded EXCEPT ![j] = FALSE] /\ attemptState' = [attemptState EXCEPT ![a] = AttemptAuthorized] /\ attemptWorker' = [attemptWorker EXCEPT ![a] = w] /\ attemptProgress' = [attemptProgress EXCEPT ![a] = IF checkpointValid[j] THEN checkpoint[j] ELSE 0] /\ initialProgress' = [initialProgress EXCEPT ![a] = IF checkpointValid[j] THEN checkpoint[j] ELSE 0] /\ resumedFromCheckpoint' = [resumedFromCheckpoint EXCEPT ![a] = checkpointValid[j] /\ checkpoint[j] > 0] /\ stageState' = [stageState EXCEPT ![a] = IF ~checkpointValid[j] \/ checkpoint[j] = 0 THEN StageNone ELSE IF checkpoint[j] = MaxProgress THEN StageComplete ELSE StagePartial] /\ UNCHANGED <> /\ UNCHANGED coordinationVars /\ UNCHANGED workerVars /\ UNCHANGED pendingOutcome ActivateAttempt(a) == LET j == JobOf(a) w == attemptWorker[a] IN /\ CurrentAuthority(a) /\ phase[j] = Running /\ attemptState[a] = AttemptAuthorized /\ reservation[j] = w /\ workerState[w] = WorkerUp /\ CanReach(w, CoordSvc) /\ reservation' = [reservation EXCEPT ![j] = NoWorker] /\ leaseState' = [leaseState EXCEPT ![a] = LeaseLive] /\ leaseAge' = [leaseAge EXCEPT ![a] = 0] /\ liveProgress' = [liveProgress EXCEPT ![j] = attemptProgress[a]] /\ liveProgressAttempt' = [liveProgressAttempt EXCEPT ![j] = a] /\ attemptState' = [attemptState EXCEPT ![a] = AttemptExecuting] /\ UNCHANGED durableVars /\ UNCHANGED <> /\ UNCHANGED workerVars /\ UNCHANGED <> RenewLease(a) == LET w == attemptWorker[a] IN /\ leaseState[a] = LeaseLive /\ leaseAge[a] > 0 /\ workerState[w] = WorkerUp /\ CanReach(w, CoordSvc) /\ leaseAge' = [leaseAge EXCEPT ![a] = 0] /\ UNCHANGED durableVars /\ UNCHANGED <> /\ UNCHANGED workerVars /\ UNCHANGED attemptVars Tick == /\ \E a \in AttemptIds : leaseState[a] = LeaseLive /\ leaseAge[a] < LeaseLimit /\ leaseAge' = [a \in AttemptIds |-> IF leaseState[a] = LeaseLive /\ leaseAge[a] < LeaseLimit THEN leaseAge[a] + 1 ELSE leaseAge[a]] /\ UNCHANGED durableVars /\ UNCHANGED <> /\ UNCHANGED workerVars /\ UNCHANGED attemptVars \* Lease expiry does not clear live progress: product may retain a stale \* (attempt, bytes) hint until overwrite, reader discard, or terminal GC. ExpireLease(a) == /\ leaseState[a] = LeaseLive /\ leaseAge[a] = LeaseLimit /\ leaseState' = [leaseState EXCEPT ![a] = LeaseExpired] /\ leaseAge' = [leaseAge EXCEPT ![a] = 0] /\ UNCHANGED durableVars /\ UNCHANGED <> /\ UNCHANGED workerVars /\ UNCHANGED attemptVars \* Full CoordinationStore wipe; durable authority and processes unchanged. LoseCoordinationState == /\ FaultsEnabled /\ CoordSvc \in FaultServices /\ coordLossCount < MaxCoordLosses /\ queued' = [j \in Jobs |-> FALSE] /\ reservation' = [j \in Jobs |-> NoWorker] /\ leaseState' = [a \in AttemptIds |-> IF leaseState[a] = LeaseLive THEN LeaseExpired ELSE leaseState[a]] /\ leaseAge' = [a \in AttemptIds |-> 0] /\ cancelNotice' = [j \in Jobs |-> FALSE] /\ liveProgress' = [j \in Jobs |-> 0] /\ liveProgressAttempt' = [j \in Jobs |-> NoAttempt] /\ coordLossCount' = coordLossCount + 1 /\ UNCHANGED durableVars /\ UNCHANGED workerVars /\ UNCHANGED attemptVars \* Re-partition allowed from recovered links until MaxPartitions is reached. DropLink(w, service) == /\ FaultsEnabled /\ service \in FaultServices /\ workerState[w] = WorkerUp /\ links[w][service] \in {LinkUp, LinkRecovered} /\ partitionCount < MaxPartitions /\ links' = [links EXCEPT ![w][service] = LinkDown] /\ partitionCount' = partitionCount + 1 /\ UNCHANGED durableVars /\ UNCHANGED coordinationVars /\ UNCHANGED workerState /\ UNCHANGED attemptVars RestoreLink(w, service) == /\ links[w][service] = LinkDown /\ links' = [links EXCEPT ![w][service] = LinkRecovered] /\ UNCHANGED durableVars /\ UNCHANGED coordinationVars /\ UNCHANGED workerState /\ UNCHANGED partitionCount /\ UNCHANGED attemptVars \* Crash does not clear coordination progress (Redis keys outlive the process). CrashWorker(w) == /\ FaultsEnabled /\ workerState[w] = WorkerUp /\ OtherCapacity(w) /\ workerState' = [workerState EXCEPT ![w] = WorkerDown] /\ attemptState' = [a \in AttemptIds |-> IF attemptWorker[a] = w /\ attemptState[a] \in {AttemptAuthorized, AttemptExecuting} THEN AttemptStopped ELSE attemptState[a]] /\ UNCHANGED durableVars /\ UNCHANGED coordinationVars /\ UNCHANGED links /\ UNCHANGED partitionCount /\ UNCHANGED <> StartIncarnation(w) == /\ workerState[w] = WorkerUnused /\ workerState' = [workerState EXCEPT ![w] = WorkerUp] /\ UNCHANGED durableVars /\ UNCHANGED coordinationVars /\ UNCHANGED links /\ UNCHANGED partitionCount /\ UNCHANGED attemptVars ReleaseReservation(j) == LET w == reservation[j] IN /\ w # NoWorker /\ (Terminal(j) \/ phase[j] = Cancelling \/ workerState[w] # WorkerUp \/ (authority[j] # NoAttempt /\ attemptState[authority[j]] # AttemptAuthorized)) /\ reservation' = [reservation EXCEPT ![j] = NoWorker] /\ UNCHANGED durableVars /\ UNCHANGED <> /\ UNCHANGED workerVars /\ UNCHANGED attemptVars \* Revoke durable authority; stop authorized-never-activated attempts so staging \* can clean. Executing attempts become zombies until CooperativeStop. \* Progress stays in coordination until ClearStaleLiveProgress (not co-stepped). ReapLostAttempt(a) == LET j == JobOf(a) IN /\ CurrentAuthority(a) /\ phase[j] = Running /\ leaseState[a] # LeaseLive /\ (attemptState[a] # AttemptAuthorized \/ reservation[j] = NoWorker) /\ phase' = [phase EXCEPT ![j] = IF currentToken[j] < MaxAttempts THEN RetryWait ELSE Failed] /\ authority' = [authority EXCEPT ![j] = NoAttempt] /\ attemptState' = [attemptState EXCEPT ![a] = IF attemptState[a] = AttemptAuthorized THEN AttemptStopped ELSE @] /\ UNCHANGED <> /\ UNCHANGED coordinationVars /\ UNCHANGED workerVars /\ UNCHANGED <> ScheduleRetry(j) == /\ phase[j] = RetryWait /\ phase' = [phase EXCEPT ![j] = Queued] /\ dispatchNeeded' = [dispatchNeeded EXCEPT ![j] = TRUE] /\ UNCHANGED <> /\ UNCHANGED coordinationVars /\ UNCHANGED workerVars /\ UNCHANGED attemptVars RequestCancel(j) == /\ phase[j] \in {Queued, Running, RetryWait} /\ cancelRequested' = [cancelRequested EXCEPT ![j] = TRUE] /\ phase' = [phase EXCEPT ![j] = IF phase[j] = Running THEN Cancelling ELSE Cancelled] /\ authority' = [authority EXCEPT ![j] = IF phase[j] = Running THEN @ ELSE NoAttempt] /\ dispatchNeeded' = [dispatchNeeded EXCEPT ![j] = FALSE] /\ UNCHANGED <> /\ UNCHANGED coordinationVars /\ UNCHANGED workerVars /\ UNCHANGED attemptVars \* Cancelled forbids durable publication; it does not assert process stop. \* Progress remains until coord GC (ClearStaleLiveProgress / terminal clean). \* Authorized-never-activated attempts stop here so they do not block quiescence. FinalizeCancel(j) == /\ phase[j] = Cancelling /\ phase' = [phase EXCEPT ![j] = Cancelled] /\ authority' = [authority EXCEPT ![j] = NoAttempt] /\ attemptState' = [a \in AttemptIds |-> IF JobOf(a) = j /\ attemptState[a] = AttemptAuthorized THEN AttemptStopped ELSE attemptState[a]] /\ UNCHANGED <> /\ UNCHANGED coordinationVars /\ UNCHANGED workerVars /\ UNCHANGED <> \* In-flight cancel signal only; terminal jobs do not re-arm notice after clean. PublishCancelNotice(j) == /\ cancelRequested[j] /\ ~Terminal(j) /\ ~cancelNotice[j] /\ cancelNotice' = [cancelNotice EXCEPT ![j] = TRUE] /\ UNCHANGED durableVars /\ UNCHANGED <> /\ UNCHANGED workerVars /\ UNCHANGED attemptVars CooperativeStop(a) == LET j == JobOf(a) w == attemptWorker[a] IN /\ attemptState[a] = AttemptExecuting /\ (leaseState[a] = LeaseExpired \/ cancelNotice[j] \/ (CanReach(w, DurableSvc) /\ cancelRequested[j]) \/ ~CurrentAuthority(a)) /\ attemptState' = [attemptState EXCEPT ![a] = AttemptStopped] /\ UNCHANGED durableVars /\ UNCHANGED coordinationVars /\ UNCHANGED workerVars /\ UNCHANGED <> \* Coord-only lease drop; progress GC is separate (ClearStaleLiveProgress). ReleaseLease(a) == LET j == JobOf(a) IN /\ leaseState[a] = LeaseLive /\ (attemptState[a] \in {AttemptStopped, AttemptCommitted} \/ ~CurrentAuthority(a) \/ Terminal(j)) /\ leaseState' = [leaseState EXCEPT ![a] = LeaseExpired] /\ leaseAge' = [leaseAge EXCEPT ![a] = 0] /\ UNCHANGED durableVars /\ UNCHANGED <> /\ UNCHANGED workerVars /\ UNCHANGED attemptVars \* Ambient: drop progress that is not backed by current authority + live lease. \* Models Redis GC / reaper independent of durable CAS. ClearStaleLiveProgress(j) == /\ liveProgressAttempt[j] # NoAttempt /\ LET a == liveProgressAttempt[j] IN ~(CurrentAuthority(a) /\ leaseState[a] = LeaseLive) /\ liveProgress' = [liveProgress EXCEPT ![j] = 0] /\ liveProgressAttempt' = [liveProgressAttempt EXCEPT ![j] = NoAttempt] /\ UNCHANGED durableVars /\ UNCHANGED <> /\ UNCHANGED workerVars /\ UNCHANGED attemptVars CleanTerminalCoordination(j) == /\ Terminal(j) /\ (queued[j] \/ reservation[j] # NoWorker \/ cancelNotice[j] \/ liveProgress[j] # 0 \/ liveProgressAttempt[j] # NoAttempt) /\ queued' = [queued EXCEPT ![j] = FALSE] /\ reservation' = [reservation EXCEPT ![j] = NoWorker] /\ cancelNotice' = [cancelNotice EXCEPT ![j] = FALSE] /\ liveProgress' = [liveProgress EXCEPT ![j] = 0] /\ liveProgressAttempt' = [liveProgressAttempt EXCEPT ![j] = NoAttempt] /\ UNCHANGED durableVars /\ UNCHANGED <> /\ UNCHANGED workerVars /\ UNCHANGED attemptVars \* Target write remains enabled after lease loss; staging identity fences publish. AdvanceAttempt(a, amount) == LET w == attemptWorker[a] IN /\ attemptState[a] = AttemptExecuting /\ workerState[w] = WorkerUp /\ CanReach(w, TargetSvc) /\ amount \in 1..MaxProgress /\ amount > attemptProgress[a] /\ attemptProgress' = [attemptProgress EXCEPT ![a] = amount] /\ stageState' = [stageState EXCEPT ![a] = IF amount = MaxProgress THEN StageComplete ELSE StagePartial] /\ UNCHANGED durableVars /\ UNCHANGED coordinationVars /\ UNCHANGED workerVars /\ UNCHANGED <> PersistCheckpoint(a) == LET j == JobOf(a) w == attemptWorker[a] IN /\ CurrentAuthority(a) /\ phase[j] = Running /\ attemptState[a] = AttemptExecuting /\ workerState[w] = WorkerUp /\ CanReach(w, DurableSvc) /\ attemptProgress[a] > 0 /\ (~checkpointValid[j] \/ checkpoint[j] # attemptProgress[a]) /\ checkpoint' = [checkpoint EXCEPT ![j] = attemptProgress[a]] /\ checkpointValid' = [checkpointValid EXCEPT ![j] = TRUE] /\ UNCHANGED <> /\ UNCHANGED coordinationVars /\ UNCHANGED workerVars /\ UNCHANGED attemptVars InvalidateCheckpoint(j) == /\ ~Terminal(j) /\ checkpointValid[j] /\ checkpointValid' = [checkpointValid EXCEPT ![j] = FALSE] /\ UNCHANGED <> /\ UNCHANGED coordinationVars /\ UNCHANGED workerVars /\ UNCHANGED attemptVars ReportProgress(a) == LET j == JobOf(a) w == attemptWorker[a] IN /\ CurrentAuthority(a) /\ phase[j] = Running /\ leaseState[a] = LeaseLive /\ attemptState[a] = AttemptExecuting /\ workerState[w] = WorkerUp /\ CanReach(w, CoordSvc) /\ (liveProgress[j] # attemptProgress[a] \/ liveProgressAttempt[j] # a) /\ liveProgress' = [liveProgress EXCEPT ![j] = attemptProgress[a]] /\ liveProgressAttempt' = [liveProgressAttempt EXCEPT ![j] = a] /\ UNCHANGED durableVars /\ UNCHANGED <> /\ UNCHANGED workerVars /\ UNCHANGED attemptVars SendOutcome(a, outcome) == LET w == attemptWorker[a] IN /\ outcome \in {WantsSuccess} \cup FailureOutcomes /\ attemptState[a] = AttemptExecuting /\ pendingOutcome[a] = NoOutcome /\ workerState[w] = WorkerUp /\ CanReach(w, DurableSvc) /\ (outcome \in FailureOutcomes \/ stageState[a] = StageComplete) /\ pendingOutcome' = [pendingOutcome EXCEPT ![a] = outcome] /\ UNCHANGED durableVars /\ UNCHANGED coordinationVars /\ UNCHANGED workerVars /\ UNCHANGED <> CommitEligible(a) == LET j == JobOf(a) IN /\ pendingOutcome[a] = WantsSuccess /\ CurrentAuthority(a) /\ phase[j] = Running /\ ~cancelRequested[j] /\ stageState[a] = StageComplete FailureEligible(a) == LET j == JobOf(a) IN /\ pendingOutcome[a] \in FailureOutcomes /\ CurrentAuthority(a) /\ phase[j] = Running /\ ~cancelRequested[j] \* Sole publication linearization point: durable fenced CAS. \* Does not touch coordination progress (cleared by ambient coord GC). AcceptSuccess(a) == LET j == JobOf(a) IN /\ CommitEligible(a) /\ phase' = [phase EXCEPT ![j] = Succeeded] /\ authority' = [authority EXCEPT ![j] = NoAttempt] /\ committedAttempt' = [committedAttempt EXCEPT ![j] = a] /\ checkpoint' = [checkpoint EXCEPT ![j] = MaxProgress] /\ checkpointValid' = [checkpointValid EXCEPT ![j] = TRUE] /\ attemptState' = [attemptState EXCEPT ![a] = AttemptCommitted] /\ pendingOutcome' = [pendingOutcome EXCEPT ![a] = NoOutcome] /\ UNCHANGED <> /\ UNCHANGED coordinationVars /\ UNCHANGED workerVars /\ UNCHANGED <> AcceptFailure(a, outcome) == LET j == JobOf(a) IN /\ pendingOutcome[a] = outcome /\ FailureEligible(a) /\ phase' = [phase EXCEPT ![j] = IF outcome = WantsPermanentFailure THEN Failed ELSE IF currentToken[j] < MaxAttempts THEN RetryWait ELSE Failed] /\ authority' = [authority EXCEPT ![j] = NoAttempt] /\ attemptState' = [attemptState EXCEPT ![a] = AttemptStopped] /\ pendingOutcome' = [pendingOutcome EXCEPT ![a] = NoOutcome] /\ UNCHANGED <> /\ UNCHANGED coordinationVars /\ UNCHANGED workerVars /\ UNCHANGED <> AcceptPermanentFailure(a) == /\ pendingOutcome[a] = WantsPermanentFailure /\ AcceptFailure(a, WantsPermanentFailure) AcceptRetryableFailureWithBudget(a) == /\ currentToken[JobOf(a)] < MaxAttempts /\ AcceptFailure(a, WantsRetryableFailure) AcceptRetryableFailureExhausted(a) == /\ currentToken[JobOf(a)] = MaxAttempts /\ AcceptFailure(a, WantsRetryableFailure) RejectOutcome(a) == /\ pendingOutcome[a] # NoOutcome /\ ~CommitEligible(a) /\ ~FailureEligible(a) /\ pendingOutcome' = [pendingOutcome EXCEPT ![a] = NoOutcome] /\ UNCHANGED durableVars /\ UNCHANGED coordinationVars /\ UNCHANGED workerVars /\ UNCHANGED <> CleanupStage(a) == LET j == JobOf(a) IN /\ stageState[a] \in {StagePartial, StageComplete} /\ committedAttempt[j] # a /\ ~CurrentAuthority(a) /\ (attemptState[a] = AttemptStopped \/ workerState[attemptWorker[a]] = WorkerDown) /\ stageState' = [stageState EXCEPT ![a] = StageCleaned] /\ UNCHANGED durableVars /\ UNCHANGED coordinationVars /\ UNCHANGED workerVars /\ UNCHANGED <> \* Stuttering sink for CHECK_DEADLOCK when SystemQuiescent; not a product API. Done == /\ SystemQuiescent /\ UNCHANGED vars ReserveJob(j) == \E w \in Workers : Reserve(j, w) AuthorizeJob(j) == \E w \in Workers : AuthorizeAttempt(j, w) ActivateJob(j) == \E a \in AttemptIds : JobOf(a) = j /\ ActivateAttempt(a) ExecuteJob(j) == \E a \in AttemptIds : /\ JobOf(a) = j /\ (SendOutcome(a, WantsRetryableFailure) \/ SendOutcome(a, WantsPermanentFailure) \/ SendOutcome(a, WantsSuccess) \/ \E amount \in 1..MaxProgress : AdvanceAttempt(a, amount)) ProcessOutcome(a) == \/ AcceptSuccess(a) \/ AcceptPermanentFailure(a) \/ AcceptRetryableFailureWithBudget(a) \/ AcceptRetryableFailureExhausted(a) \/ RejectOutcome(a) ReapJob(j) == \E a \in AttemptIds : JobOf(a) = j /\ ReapLostAttempt(a) Next == \/ \E j \in Jobs : Enqueue(j) \/ \E j \in Jobs, w \in Workers : Reserve(j, w) \/ \E j \in Jobs, w \in Workers : AuthorizeAttempt(j, w) \/ \E a \in AttemptIds : ActivateAttempt(a) \/ \E a \in AttemptIds : RenewLease(a) \/ Tick \/ \E a \in AttemptIds : ExpireLease(a) \/ LoseCoordinationState \/ \E w \in Workers, s \in Services : DropLink(w, s) \/ \E w \in Workers, s \in Services : RestoreLink(w, s) \/ \E w \in Workers : CrashWorker(w) \/ \E w \in Workers : StartIncarnation(w) \/ \E j \in Jobs : ReleaseReservation(j) \/ \E a \in AttemptIds : ReapLostAttempt(a) \/ \E j \in Jobs : ScheduleRetry(j) \/ \E j \in Jobs : RequestCancel(j) \/ \E j \in Jobs : FinalizeCancel(j) \/ \E j \in Jobs : PublishCancelNotice(j) \/ \E a \in AttemptIds : CooperativeStop(a) \/ \E a \in AttemptIds : ReleaseLease(a) \/ \E j \in Jobs : ClearStaleLiveProgress(j) \/ \E j \in Jobs : CleanTerminalCoordination(j) \/ \E a \in AttemptIds, n \in 1..MaxProgress : AdvanceAttempt(a, n) \/ \E a \in AttemptIds : PersistCheckpoint(a) \/ \E j \in Jobs : InvalidateCheckpoint(j) \/ \E a \in AttemptIds : ReportProgress(a) \/ \E a \in AttemptIds, o \in {WantsSuccess} \cup FailureOutcomes : SendOutcome(a, o) \/ \E a \in AttemptIds : AcceptSuccess(a) \/ \E a \in AttemptIds : AcceptPermanentFailure(a) \/ \E a \in AttemptIds : AcceptRetryableFailureWithBudget(a) \/ \E a \in AttemptIds : AcceptRetryableFailureExhausted(a) \/ \E a \in AttemptIds : RejectOutcome(a) \/ \E a \in AttemptIds : CleanupStage(a) \/ Done \* Protocol-agent fairness is separate from RecoveryAssumption below. Fairness == /\ \A j \in Jobs : WF_vars(Enqueue(j)) /\ \A j \in Jobs : SF_vars(ReserveJob(j)) /\ \A j \in Jobs : SF_vars(AuthorizeJob(j)) /\ \A j \in Jobs : SF_vars(ActivateJob(j)) /\ \A j \in Jobs : SF_vars(ExecuteJob(j)) /\ \A a \in AttemptIds : WF_vars(ProcessOutcome(a)) /\ \A a \in AttemptIds : WF_vars(ExpireLease(a)) /\ \A a \in AttemptIds : WF_vars(CooperativeStop(a)) /\ \A a \in AttemptIds : WF_vars(ReleaseLease(a)) /\ \A a \in AttemptIds : WF_vars(CleanupStage(a)) /\ \A j \in Jobs : WF_vars(ReleaseReservation(j)) /\ \A j \in Jobs : WF_vars(ReapJob(j)) /\ \A j \in Jobs : WF_vars(ScheduleRetry(j)) /\ \A j \in Jobs : WF_vars(FinalizeCancel(j)) /\ \A j \in Jobs : WF_vars(PublishCancelNotice(j)) /\ \A j \in Jobs : WF_vars(ClearStaleLiveProgress(j)) /\ \A j \in Jobs : WF_vars(CleanTerminalCoordination(j)) /\ \A w \in Workers, s \in Services : WF_vars(RestoreLink(w, s)) /\ \A w \in Workers : WF_vars(StartIncarnation(w)) /\ WF_vars(Tick) SafetySpec == Init /\ [][Next]_vars Spec == SafetySpec /\ Fairness TypeOK == /\ phase \in [Jobs -> Phases] /\ cancelRequested \in [Jobs -> BOOLEAN] /\ currentToken \in [Jobs -> 0..MaxAttempts] /\ authority \in [Jobs -> AttemptIds \cup {NoAttempt}] /\ dispatchNeeded \in [Jobs -> BOOLEAN] /\ committedAttempt \in [Jobs -> AttemptIds \cup {NoAttempt}] /\ checkpoint \in [Jobs -> 0..MaxProgress] /\ checkpointValid \in [Jobs -> BOOLEAN] /\ queued \in [Jobs -> BOOLEAN] /\ reservation \in [Jobs -> Workers \cup {NoWorker}] /\ leaseState \in [AttemptIds -> LeaseStates] /\ leaseAge \in [AttemptIds -> 0..LeaseLimit] /\ cancelNotice \in [Jobs -> BOOLEAN] /\ liveProgress \in [Jobs -> 0..MaxProgress] /\ liveProgressAttempt \in [Jobs -> AttemptIds \cup {NoAttempt}] /\ coordLossCount \in 0..MaxCoordLosses /\ workerState \in [Workers -> WorkerStates] /\ links \in [Workers -> [Services -> LinkStates]] /\ partitionCount \in 0..MaxPartitions /\ attemptState \in [AttemptIds -> AttemptStates] /\ attemptWorker \in [AttemptIds -> Workers \cup {NoWorker}] /\ attemptProgress \in [AttemptIds -> 0..MaxProgress] /\ initialProgress \in [AttemptIds -> 0..MaxProgress] /\ resumedFromCheckpoint \in [AttemptIds -> BOOLEAN] /\ stageState \in [AttemptIds -> StageStates] /\ pendingOutcome \in [AttemptIds -> OutcomeRequests] StateConsistency == /\ \A j \in Jobs : /\ (phase[j] \in {Running, Cancelling} => /\ authority[j] \in AttemptIds /\ JobOf(authority[j]) = j /\ TokenOf(authority[j]) = currentToken[j]) /\ (phase[j] \in {Queued, RetryWait} \cup TerminalPhases => authority[j] = NoAttempt) /\ (phase[j] \in {Queued, Running, RetryWait} => ~cancelRequested[j]) /\ (phase[j] \in {Cancelling, Cancelled} => cancelRequested[j]) /\ (phase[j] \in {Succeeded, Failed} => ~cancelRequested[j]) /\ (committedAttempt[j] # NoAttempt <=> phase[j] = Succeeded) /\ (phase[j] = Queued => dispatchNeeded[j]) /\ (reservation[j] # NoWorker => ~queued[j]) /\ \A a \in AttemptIds : /\ (attemptState[a] = AttemptUnused <=> attemptWorker[a] = NoWorker) /\ (attemptState[a] # AttemptUnused => /\ attemptWorker[a] \in Workers /\ TokenOf(a) <= currentToken[JobOf(a)]) /\ (leaseState[a] = LeaseLive => attemptState[a] # AttemptUnused) /\ (stageState[a] = StageCleaned => committedAttempt[JobOf(a)] # a) \* Tokens 1..currentToken are allocated densely (claim linearization). AllocatedTokensUsed == \A j \in Jobs : \A t \in 1..currentToken[j] : attemptState[AttemptId(j, t)] # AttemptUnused \* Live progress is a coordination hint, not a cross-store snapshot invariant. \* Stale tuples may remain after lease loss; readers revalidate against authority. LiveProgressWellFormed == \A j \in Jobs : liveProgressAttempt[j] # NoAttempt => LET a == liveProgressAttempt[j] IN /\ JobOf(a) = j /\ attemptState[a] # AttemptUnused /\ TokenOf(a) <= currentToken[j] /\ liveProgress[j] <= attemptProgress[a] \* Catches leftover zombie progress under a live lease for a different attempt \* (forgot to overwrite/clear on activate/replace). Empty progress is allowed. LiveProgressCompatibleWithLiveLease == \A j \in Jobs : \A a \in AttemptIds : (JobOf(a) = j /\ leaseState[a] = LeaseLive) => liveProgressAttempt[j] \in {NoAttempt, a} \* At most one deployment-wide live lease per job. AtMostOneLiveLeasePerJob == \A j \in Jobs : Cardinality({a \in AttemptIds : JobOf(a) = j /\ leaseState[a] = LeaseLive}) <= 1 \* Authority, if present, is a non-unused attempt of this job at currentToken. AuthorityWellFormed == \A j \in Jobs : authority[j] # NoAttempt => /\ JobOf(authority[j]) = j /\ TokenOf(authority[j]) = currentToken[j] /\ attemptState[authority[j]] \in {AttemptAuthorized, AttemptExecuting, AttemptStopped} CommittedResultConsistent == \A j \in Jobs : committedAttempt[j] # NoAttempt => /\ JobOf(committedAttempt[j]) = j /\ TokenOf(committedAttempt[j]) = currentToken[j] /\ attemptState[committedAttempt[j]] = AttemptCommitted /\ stageState[committedAttempt[j]] = StageComplete /\ ~cancelRequested[j] CancellationSafety == \A j \in Jobs : cancelRequested[j] => committedAttempt[j] = NoAttempt PermitSafety == /\ Cardinality(ReservedJobs) + Cardinality(LiveLeases) <= GlobalLimit /\ \A w \in Workers : Cardinality(ReservedFor(w)) + Cardinality(ExecutingFor(w)) <= LocalLimit ResumeValidity == \A a \in AttemptIds : /\ attemptProgress[a] >= initialProgress[a] /\ (initialProgress[a] > 0 => resumedFromCheckpoint[a]) /\ (resumedFromCheckpoint[a] => initialProgress[a] > 0) \* Uncommitted staging is only retained for live or stopped attempts, not cleaned \* while still authoritative (cleanup gated in CleanupStage). StagingOwnership == \A a \in AttemptIds : /\ (stageState[a] = StageCleaned => /\ committedAttempt[JobOf(a)] # a /\ ~CurrentAuthority(a)) /\ (committedAttempt[JobOf(a)] = a => stageState[a] = StageComplete) TerminalStep == \A j \in Jobs : Terminal(j) => /\ phase'[j] = phase[j] /\ cancelRequested'[j] = cancelRequested[j] /\ currentToken'[j] = currentToken[j] /\ authority'[j] = authority[j] /\ committedAttempt'[j] = committedAttempt[j] /\ checkpoint'[j] = checkpoint[j] /\ checkpointValid'[j] = checkpointValid[j] FenceStep == \A j \in Jobs : currentToken'[j] >= currentToken[j] CommitLinearizationStep == \A j \in Jobs : committedAttempt'[j] # committedAttempt[j] => /\ committedAttempt[j] = NoAttempt /\ phase[j] = Running /\ authority[j] = committedAttempt'[j] /\ TokenOf(committedAttempt'[j]) = currentToken[j] /\ pendingOutcome[committedAttempt'[j]] = WantsSuccess /\ ~cancelRequested[j] /\ stageState[committedAttempt'[j]] = StageComplete FailureDispositionStep == \A a \in AttemptIds : LET j == JobOf(a) IN /\ FailureEligible(a) /\ attemptState'[a] = AttemptStopped /\ pendingOutcome'[a] = NoOutcome /\ authority'[j] = NoAttempt => phase'[j] = IF pendingOutcome[a] = WantsPermanentFailure THEN Failed ELSE IF currentToken[j] < MaxAttempts THEN RetryWait ELSE Failed TerminalStateIsImmutable == [][TerminalStep]_vars FencingTokensMonotonic == [][FenceStep]_vars CommitIsFencedCAS == [][CommitLinearizationStep]_vars FailureDispositionHonored == [][FailureDispositionStep]_vars \* Liveness is conditional on eventual lasting capacity. Safety needs no recovery. \* WorkerAvailable requires Coord, Durable, and Target reachable for some Up worker. \* EventuallyDispatched: accepted work leaves Queued (starts Running or ends terminal, \* including Cancelled). It is not "successful delivery of bytes." \* EventuallyTerminal: some terminal outcome (Succeeded, Failed, or Cancelled). \* Permanent failure may yield Failed immediately; retryable failure may yield \* Failed after the retry budget. Success is not guaranteed under either choice. \* EventuallyNonWinnerStagingCleaned / EventuallySystemQuiescent: model GC goal under \* fairness (infallible CleanupStage). Product seeks zero orphans under recovery; \* not a safety invariant and not a readiness gate (IF-7). WorkerAvailable(w) == /\ workerState[w] = WorkerUp /\ \A s \in Services : CanReach(w, s) RecoveryAssumption == <>[](\E w \in Workers : WorkerAvailable(w)) EventuallyDispatched == RecoveryAssumption => \A j \in Jobs : phase[j] = Queued ~> (phase[j] = Running \/ Terminal(j)) EventuallyReaped == RecoveryAssumption => \A a \in AttemptIds : (CurrentAuthority(a) /\ leaseState[a] # LeaseLive) ~> (~CurrentAuthority(a) \/ Terminal(JobOf(a))) EventuallyTerminal == RecoveryAssumption => \A j \in Jobs : <>Terminal(j) CleanupEligible(a) == /\ stageState[a] \in {StagePartial, StageComplete} /\ committedAttempt[JobOf(a)] # a /\ ~CurrentAuthority(a) /\ (attemptState[a] = AttemptStopped \/ workerState[attemptWorker[a]] = WorkerDown) EventuallyNonWinnerStagingCleaned == RecoveryAssumption => \A a \in AttemptIds : CleanupEligible(a) ~> stageState[a] = StageCleaned EventuallySystemQuiescent == RecoveryAssumption => <>SystemQuiescent =============================================================================