------------------------------ MODULE OdmPipeline ------------------------------
(***************************************************************************)
(* A durable workflow engine driving a multi-stage pipeline with           *)
(* at-least-once stage execution over shared on-disk outputs.              *)
(*                                                                         *)
(* Each stage's activity may crash mid-write, leaving Partial outputs on   *)
(* disk.  The stage runner skips any stage whose outputs already exist     *)
(* (so completed stages are never redone).  The design rule under test:    *)
(* a RETRY of a crashed stage must force a from-scratch re-run of that     *)
(* stage (ForceRerunOnRetry = TRUE, i.e. `--rerun-from <stage>`).          *)
(* With ForceRerunOnRetry = FALSE the retry trusts the Partial outputs,    *)
(* skips the stage, and NeverTrustPartial is violated.                     *)
(*                                                                         *)
(* Also modeled: workflow-ID uniqueness (a duplicate submit for the same   *)
(* dataset is rejected while a run is active) and mandatory cleanup        *)
(* before every terminal state, including failure and cancellation.       *)
(***************************************************************************)
EXTENDS Naturals

CONSTANTS
    NumStages,          \* number of sequential pipeline stages
    MaxAttempts,        \* per-stage attempt bound (keeps the model finite)
    ForceRerunOnRetry   \* TRUE: retries re-run the stage from scratch

ASSUME NumStages \in Nat \ {0}
ASSUME MaxAttempts \in Nat \ {0}
ASSUME ForceRerunOnRetry \in BOOLEAN

Stages      == 1..NumStages
StageStates == {"NotStarted", "Partial", "Complete"}
Terminals   == {"Done", "Failed", "Cancelled"}

VARIABLES
    phase,      \* Idle | Staging | Running | Uploading | CleaningUp | terminal
    cur,        \* index of the stage being run (0 when phase # "Running")
    stages,     \* on-disk output state per stage
    attempts,   \* per-stage activity attempt count
    outcome,    \* terminal state pending after cleanup
    released,   \* TRUE once the project dir has been cleaned up
    activeRuns, \* runs currently active for the one dataset
    pending     \* submit requests not yet accepted or rejected

vars == <<phase, cur, stages, attempts, outcome, released, activeRuns, pending>>

TypeOK ==
    /\ phase \in {"Idle", "Staging", "Running", "Uploading", "CleaningUp"}
               \cup Terminals
    /\ cur \in 0..NumStages
    /\ stages \in [Stages -> StageStates]
    /\ attempts \in [Stages -> 0..MaxAttempts]
    /\ outcome \in {"None"} \cup Terminals
    /\ released \in BOOLEAN
    /\ activeRuns \in 0..2
    /\ pending \in 0..2

Init ==
    /\ phase = "Idle"
    /\ cur = 0
    /\ stages = [i \in Stages |-> "NotStarted"]
    /\ attempts = [i \in Stages |-> 0]
    /\ outcome = "None"
    /\ released = FALSE
    /\ activeRuns = 0
    /\ pending = 2

(* Workflow-ID uniqueness: a submit starts a run only when none is active. *)
Submit ==
    /\ phase = "Idle" /\ pending > 0 /\ activeRuns = 0
    /\ pending' = pending - 1
    /\ activeRuns' = 1
    /\ phase' = "Staging"
    /\ UNCHANGED <<cur, stages, attempts, outcome, released>>

(* A duplicate submit while a run is active is rejected outright. *)
RejectDuplicate ==
    /\ pending > 0 /\ activeRuns = 1
    /\ pending' = pending - 1
    /\ UNCHANGED <<phase, cur, stages, attempts, outcome, released, activeRuns>>

StagingDone ==
    /\ phase = "Staging"
    /\ phase' = "Running" /\ cur' = 1
    /\ UNCHANGED <<stages, attempts, outcome, released, activeRuns, pending>>

Advance(i) ==
    IF i = NumStages
        THEN phase' = "Uploading" /\ cur' = 0
        ELSE phase' = phase       /\ cur' = i + 1

(* A retry of a Partial stage really executes it only under rerun-from;    *)
(* without it the runner would skip, so execution (and crashing) is only   *)
(* possible when the stage has no partial outputs or rerun is forced.      *)
CanExecute(i) == stages[i] = "Partial" => ForceRerunOnRetry

StageCompletes ==
    /\ phase = "Running"
    /\ attempts[cur] < MaxAttempts
    /\ CanExecute(cur)
    /\ attempts' = [attempts EXCEPT ![cur] = @ + 1]
    /\ stages' = [stages EXCEPT ![cur] = "Complete"]
    /\ Advance(cur)
    /\ UNCHANGED <<outcome, released, activeRuns, pending>>

StageCrashes ==
    /\ phase = "Running"
    /\ attempts[cur] < MaxAttempts
    /\ CanExecute(cur)
    /\ attempts' = [attempts EXCEPT ![cur] = @ + 1]
    /\ stages' = [stages EXCEPT ![cur] = "Partial"]
    /\ UNCHANGED <<phase, cur, outcome, released, activeRuns, pending>>

(* Flawed design only: the retry sees outputs on disk, trusts them, and    *)
(* skips the stage even though they are Partial.                           *)
SkipPartialAsComplete ==
    /\ phase = "Running"
    /\ ~ForceRerunOnRetry
    /\ stages[cur] = "Partial"
    /\ attempts[cur] < MaxAttempts
    /\ attempts' = [attempts EXCEPT ![cur] = @ + 1]
    /\ Advance(cur)
    /\ UNCHANGED <<stages, outcome, released, activeRuns, pending>>

StageExhausted ==
    /\ phase = "Running"
    /\ attempts[cur] = MaxAttempts
    /\ stages[cur] # "Complete"
    /\ phase' = "CleaningUp" /\ outcome' = "Failed" /\ cur' = 0
    /\ UNCHANGED <<stages, attempts, released, activeRuns, pending>>

(* Cancellation can arrive during any running phase; cleanup still runs. *)
Cancel ==
    /\ phase \in {"Staging", "Running", "Uploading"}
    /\ phase' = "CleaningUp" /\ outcome' = "Cancelled" /\ cur' = 0
    /\ UNCHANGED <<stages, attempts, released, activeRuns, pending>>

UploadDone ==
    /\ phase = "Uploading"
    /\ phase' = "CleaningUp" /\ outcome' = "Done"
    /\ UNCHANGED <<cur, stages, attempts, released, activeRuns, pending>>

CleanupDone ==
    /\ phase = "CleaningUp"
    /\ released' = TRUE
    /\ phase' = outcome
    /\ activeRuns' = 0
    /\ UNCHANGED <<cur, stages, attempts, outcome, pending>>

(* Terminal states stutter so TLC reports no spurious deadlock. *)
Terminal ==
    /\ phase \in Terminals
    /\ UNCHANGED vars

Next ==
    \/ Submit
    \/ RejectDuplicate
    \/ StagingDone
    \/ StageCompletes
    \/ StageCrashes
    \/ SkipPartialAsComplete
    \/ StageExhausted
    \/ Cancel
    \/ UploadDone
    \/ CleanupDone
    \/ Terminal

Spec == Init /\ [][Next]_vars

--------------------------------------------------------------------------------
(* Invariants *)

(* The workflow never advances past a stage whose outputs are Partial. *)
NeverTrustPartial ==
    \A i \in Stages :
        stages[i] = "Partial" =>
            /\ phase \notin {"Uploading", "Done"}
            /\ ~(phase = "Running" /\ cur > i)
            /\ ~(phase = "CleaningUp" /\ outcome = "Done")

(* At most one run is ever active for the dataset. *)
NoConcurrentRuns == activeRuns <= 1

(* A terminal state is reached only after the project dir was released. *)
CleanupAlways == phase \in Terminals => released

(* A Complete stage stays Complete for the remainder of the run. *)
MonotonicProgress ==
    [][\A i \in Stages :
          stages[i] = "Complete" => stages'[i] = "Complete"]_vars

================================================================================