1000 lines
40 KiB
Text
1000 lines
40 KiB
Text
------------------------------ 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"
|
|
WantsFailure == "WantsFailure"
|
|
OutcomeRequests == {NoOutcome, WantsSuccess, WantsFailure}
|
|
|
|
NoAttempt == [job |-> NoJob, token |-> 0]
|
|
|
|
\* Product: opaque attempt id + separate fence; model fuses them as <<job, token>>.
|
|
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 ==
|
|
<<phase, cancelRequested, currentToken, authority, dispatchNeeded,
|
|
committedAttempt, checkpoint, checkpointValid>>
|
|
coordinationVars ==
|
|
<<queued, reservation, leaseState, leaseAge, cancelNotice,
|
|
liveProgress, liveProgressAttempt, coordLossCount>>
|
|
workerVars == <<workerState, links, partitionCount>>
|
|
attemptVars ==
|
|
<<attemptState, attemptWorker, attemptProgress, initialProgress,
|
|
resumedFromCheckpoint, stageState, pendingOutcome>>
|
|
vars == <<durableVars, coordinationVars, workerVars, attemptVars>>
|
|
|
|
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 <<reservation, leaseState, leaseAge, cancelNotice,
|
|
liveProgress, liveProgressAttempt, coordLossCount>>
|
|
/\ 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 <<leaseState, leaseAge, cancelNotice, liveProgress,
|
|
liveProgressAttempt, coordLossCount>>
|
|
/\ 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 <<cancelRequested, committedAttempt, checkpoint,
|
|
checkpointValid>>
|
|
/\ 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 <<queued, cancelNotice, coordLossCount>>
|
|
/\ UNCHANGED workerVars
|
|
/\ UNCHANGED <<attemptWorker, attemptProgress, initialProgress,
|
|
resumedFromCheckpoint, stageState, pendingOutcome>>
|
|
|
|
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 <<queued, reservation, leaseState, cancelNotice,
|
|
liveProgress, liveProgressAttempt, coordLossCount>>
|
|
/\ 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 <<queued, reservation, leaseState, cancelNotice,
|
|
liveProgress, liveProgressAttempt, coordLossCount>>
|
|
/\ 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 <<queued, reservation, cancelNotice, liveProgress,
|
|
liveProgressAttempt, coordLossCount>>
|
|
/\ 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 <<attemptWorker, attemptProgress, initialProgress,
|
|
resumedFromCheckpoint, stageState, pendingOutcome>>
|
|
|
|
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 <<queued, leaseState, leaseAge, cancelNotice,
|
|
liveProgress, liveProgressAttempt, coordLossCount>>
|
|
/\ 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 <<cancelRequested, currentToken, dispatchNeeded,
|
|
committedAttempt, checkpoint, checkpointValid>>
|
|
/\ UNCHANGED coordinationVars
|
|
/\ UNCHANGED workerVars
|
|
/\ UNCHANGED <<attemptWorker, attemptProgress, initialProgress,
|
|
resumedFromCheckpoint, stageState, pendingOutcome>>
|
|
|
|
ScheduleRetry(j) ==
|
|
/\ phase[j] = RetryWait
|
|
/\ phase' = [phase EXCEPT ![j] = Queued]
|
|
/\ dispatchNeeded' = [dispatchNeeded EXCEPT ![j] = TRUE]
|
|
/\ UNCHANGED <<cancelRequested, currentToken, authority,
|
|
committedAttempt, checkpoint, checkpointValid>>
|
|
/\ 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 <<currentToken, committedAttempt, checkpoint, checkpointValid>>
|
|
/\ 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 <<cancelRequested, currentToken, dispatchNeeded,
|
|
committedAttempt, checkpoint, checkpointValid>>
|
|
/\ UNCHANGED coordinationVars
|
|
/\ UNCHANGED workerVars
|
|
/\ UNCHANGED <<attemptWorker, attemptProgress, initialProgress,
|
|
resumedFromCheckpoint, stageState, pendingOutcome>>
|
|
|
|
\* 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 <<queued, reservation, leaseState, leaseAge, liveProgress,
|
|
liveProgressAttempt, coordLossCount>>
|
|
/\ 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 <<attemptWorker, attemptProgress, initialProgress,
|
|
resumedFromCheckpoint, stageState, pendingOutcome>>
|
|
|
|
\* 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 <<queued, reservation, cancelNotice, liveProgress,
|
|
liveProgressAttempt, coordLossCount>>
|
|
/\ 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 <<queued, reservation, leaseState, leaseAge,
|
|
cancelNotice, coordLossCount>>
|
|
/\ 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 <<leaseState, leaseAge, coordLossCount>>
|
|
/\ 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 <<attemptState, attemptWorker, initialProgress,
|
|
resumedFromCheckpoint, pendingOutcome>>
|
|
|
|
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 <<phase, cancelRequested, currentToken, authority,
|
|
dispatchNeeded, committedAttempt>>
|
|
/\ UNCHANGED coordinationVars
|
|
/\ UNCHANGED workerVars
|
|
/\ UNCHANGED attemptVars
|
|
|
|
InvalidateCheckpoint(j) ==
|
|
/\ ~Terminal(j)
|
|
/\ checkpointValid[j]
|
|
/\ checkpointValid' = [checkpointValid EXCEPT ![j] = FALSE]
|
|
/\ UNCHANGED <<phase, cancelRequested, currentToken, authority,
|
|
dispatchNeeded, committedAttempt, checkpoint>>
|
|
/\ 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 <<queued, reservation, leaseState, leaseAge,
|
|
cancelNotice, coordLossCount>>
|
|
/\ UNCHANGED workerVars
|
|
/\ UNCHANGED attemptVars
|
|
|
|
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 <<attemptState, attemptWorker, attemptProgress,
|
|
initialProgress, resumedFromCheckpoint, stageState>>
|
|
|
|
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: 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 <<cancelRequested, currentToken, dispatchNeeded>>
|
|
/\ UNCHANGED coordinationVars
|
|
/\ UNCHANGED workerVars
|
|
/\ UNCHANGED <<attemptWorker, attemptProgress, initialProgress,
|
|
resumedFromCheckpoint, stageState>>
|
|
|
|
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 <<cancelRequested, currentToken, dispatchNeeded,
|
|
committedAttempt, checkpoint, checkpointValid>>
|
|
/\ UNCHANGED coordinationVars
|
|
/\ UNCHANGED workerVars
|
|
/\ UNCHANGED <<attemptWorker, attemptProgress, initialProgress,
|
|
resumedFromCheckpoint, stageState>>
|
|
|
|
RejectOutcome(a) ==
|
|
/\ pendingOutcome[a] # NoOutcome
|
|
/\ ~CommitEligible(a)
|
|
/\ ~FailureEligible(a)
|
|
/\ pendingOutcome' = [pendingOutcome EXCEPT ![a] = NoOutcome]
|
|
/\ UNCHANGED durableVars
|
|
/\ UNCHANGED coordinationVars
|
|
/\ UNCHANGED workerVars
|
|
/\ UNCHANGED <<attemptState, attemptWorker, attemptProgress,
|
|
initialProgress, resumedFromCheckpoint, stageState>>
|
|
|
|
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 <<attemptState, attemptWorker, attemptProgress,
|
|
initialProgress, resumedFromCheckpoint, pendingOutcome>>
|
|
|
|
\* 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, 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 : 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, 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)
|
|
\/ 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
|
|
|
|
TerminalStateIsImmutable == [][TerminalStep]_vars
|
|
FencingTokensMonotonic == [][FenceStep]_vars
|
|
CommitIsFencedCAS == [][CommitLinearizationStep]_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 source/target failure may yield Failed after the retry budget; success
|
|
\* is not guaranteed under unrestricted WantsFailure nondeterminism.
|
|
\* 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
|
|
|
|
=============================================================================
|