------------------------------ MODULE Remora ------------------------------ EXTENDS Naturals, FiniteSets, TLC \* Implementation-level distributed job protocol. \* \* Durable variables are authoritative PostgreSQL state. Coordination \* variables are reconstructible Dragonfly state. Attempts remain observable \* after losing authority, so delayed requests and zombie target writes are \* checked. Target writes use an attempt-specific staging identity; only the \* durable AcceptSuccess compare-and-swap publishes a result. CONSTANTS Jobs, Workers, InitialWorkers, NoJob, NoWorker, MaxAttempts, MaxProgress, LeaseLimit, LocalLimit, GlobalLimit, FaultsEnabled, FaultServices 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"} /\ 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" WantsFailure == "WantsFailure" OutcomeRequests == {NoOutcome, WantsSuccess, WantsFailure} NoAttempt == [job |-> NoJob, token |-> 0] \* Implementations use opaque attempt IDs. The bounded model canonically uses \* <>; the token is strictly increasing and never reused per job. 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 \* Authoritative durable state. phase, cancelRequested, currentToken, authority, dispatchNeeded, committedAttempt, checkpoint, checkpointValid, \* Reconstructible coordination state. queued, reservation, leaseState, leaseAge, cancelNotice, liveProgress, liveProgressAttempt, coordinationLostOnce, \* Worker incarnations and service-specific connectivity. workerState, links, \* Explicit attempts, staging, and 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 Init == \* Jobs denotes accepted durable submissions. The outbox precedes queueing. /\ 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] /\ coordinationLostOnce = FALSE /\ workerState = [w \in Workers |-> IF w \in InitialWorkers THEN WorkerUp ELSE WorkerUnused] /\ links = [w \in Workers |-> [s \in Services |-> LinkUp]] /\ 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] \* Durable outbox projection. Duplicate dispatch is idempotent. Enqueue(j) == /\ phase[j] = Queued /\ dispatchNeeded[j] /\ ~queued[j] /\ reservation[j] = NoWorker /\ queued' = [queued EXCEPT ![j] = TRUE] /\ UNCHANGED durableVars /\ UNCHANGED <> /\ UNCHANGED workerVars /\ UNCHANGED attemptVars \* Coordination atomically reserves a logical permit. It is not authority. 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: allocate the next attempt and token. This claim linearization \* point consumes retry budget even if the worker is lost before execution. 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 \* Coordination converts the reservation to a renewable lease and execution. 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 \* Real durations refine this bounded logical age. 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 ExpireLease(a) == LET j == JobOf(a) IN /\ leaseState[a] = LeaseLive /\ leaseAge[a] = LeaseLimit /\ leaseState' = [leaseState EXCEPT ![a] = LeaseExpired] /\ leaseAge' = [leaseAge EXCEPT ![a] = 0] /\ liveProgress' = [liveProgress EXCEPT ![j] = IF liveProgressAttempt[j] = a THEN 0 ELSE @] /\ liveProgressAttempt' = [liveProgressAttempt EXCEPT ![j] = IF @ = a THEN NoAttempt ELSE @] /\ UNCHANGED durableVars /\ UNCHANGED <> /\ UNCHANGED workerVars /\ UNCHANGED attemptVars \* Failover erases ephemeral state but leaves durable state and processes. LoseCoordinationState == /\ FaultsEnabled /\ CoordSvc \in FaultServices /\ ~coordinationLostOnce /\ 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] /\ coordinationLostOnce' = TRUE /\ UNCHANGED durableVars /\ UNCHANGED workerVars /\ UNCHANGED attemptVars DropLink(w, service) == /\ FaultsEnabled /\ service \in FaultServices /\ workerState[w] = WorkerUp /\ links[w][service] = LinkUp /\ links' = [links EXCEPT ![w][service] = LinkDown] /\ 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 attemptVars CrashWorker(w) == /\ FaultsEnabled /\ workerState[w] = WorkerUp /\ 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 <> \* An unused element denotes a fresh incarnation or scale-up, never reuse. StartIncarnation(w) == /\ workerState[w] = WorkerUnused /\ workerState' = [workerState EXCEPT ![w] = WorkerUp] /\ UNCHANGED durableVars /\ UNCHANGED coordinationVars /\ UNCHANGED links /\ 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 \* Reaping revokes durable authority but does not stop a zombie process. 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] /\ UNCHANGED <> /\ UNCHANGED coordinationVars /\ UNCHANGED workerVars /\ UNCHANGED attemptVars 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 means durable publication is forbidden, not process quiescence. FinalizeCancel(j) == /\ phase[j] = Cancelling /\ phase' = [phase EXCEPT ![j] = Cancelled] /\ authority' = [authority EXCEPT ![j] = NoAttempt] /\ UNCHANGED <> /\ UNCHANGED coordinationVars /\ UNCHANGED workerVars /\ UNCHANGED attemptVars PublishCancelNotice(j) == /\ cancelRequested[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 <> 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] /\ liveProgress' = [liveProgress EXCEPT ![j] = IF liveProgressAttempt[j] = a THEN 0 ELSE @] /\ liveProgressAttempt' = [liveProgressAttempt EXCEPT ![j] = IF @ = a THEN NoAttempt ELSE @] /\ 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/replacement. Isolation by \* attempt identity prevents this physical zombie from publishing. 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 <> \* Only current authority can persist a validated reusable checkpoint. 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 \* Invocation and processing are split, allowing delayed stale requests. SendOutcome(a, outcome) == LET w == attemptWorker[a] IN /\ outcome \in {WantsSuccess, WantsFailure} /\ attemptState[a] = AttemptExecuting /\ pendingOutcome[a] = NoOutcome /\ workerState[w] = WorkerUp /\ CanReach(w, DurableSvc) /\ (outcome = WantsFailure \/ 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] = WantsFailure /\ CurrentAuthority(a) /\ phase[j] = Running /\ ~cancelRequested[j] \* Sole publication linearization point: a durable fenced CAS. 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) == LET j == JobOf(a) IN /\ FailureEligible(a) /\ phase' = [phase EXCEPT ![j] = 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 <> 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 <> 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, WantsFailure) \/ SendOutcome(a, WantsSuccess) \/ \E amount \in 1..MaxProgress : AdvanceAttempt(a, amount)) ProcessOutcome(a) == AcceptSuccess(a) \/ AcceptFailure(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 : 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, WantsFailure} : SendOutcome(a, o) \/ \E a \in AttemptIds : AcceptSuccess(a) \/ \E a \in AttemptIds : AcceptFailure(a) \/ \E a \in AttemptIds : RejectOutcome(a) \/ \E a \in AttemptIds : CleanupStage(a) \* Protocol-agent fairness is separate from the recovery assumption 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(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 \* State invariants 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}] /\ coordinationLostOnce \in BOOLEAN /\ workerState \in [Workers -> WorkerStates] /\ links \in [Workers -> [Services -> LinkStates]] /\ 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) AuthoritativeAttempts(j) == {a \in AttemptIds : CurrentAuthority(a) /\ JobOf(a) = j} AtMostOneAuthoritativeAttempt == \A j \in Jobs : Cardinality(AuthoritativeAttempts(j)) <= 1 FencingTokensUnique == \A a \in AttemptIds : attemptState[a] # AttemptUnused => TokenOf(a) <= currentToken[JobOf(a)] 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 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 QueueRecoverable == \A j \in Jobs : phase[j] = Queued => dispatchNeeded[j] AttemptIsolation == \A a \in AttemptIds : committedAttempt[JobOf(a)] = a => stageState[a] = StageComplete ResumeValidity == \A a \in AttemptIds : /\ attemptProgress[a] >= initialProgress[a] /\ (initialProgress[a] > 0 => resumedFromCheckpoint[a]) /\ (resumedFromCheckpoint[a] => initialProgress[a] > 0) ZombieIsolation == \A a \in AttemptIds : attemptState[a] = AttemptExecuting /\ ~CurrentAuthority(a) => committedAttempt[JobOf(a)] # a \* Step properties 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 TerminalStateIsImmutable == [][TerminalStep]_vars FencingTokensMonotonic == [][FenceStep]_vars CommitIsFencedCAS == [][CommitLinearizationStep]_vars \* Liveness is conditional; safety above needs no recovery assumption. WorkerAvailable(w) == /\ workerState[w] = WorkerUp /\ \A s \in Services : CanReach(w, s) RecoveryAssumption == <>[](\E w \in Workers : WorkerAvailable(w)) AtLeastOnceDelivery == 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) =============================================================================