-------------------------------- MODULE RoostPush --------------------------------
(***************************************************************************)
(* A resumable server-sent change feed over an append-only log with       *)
(* retention.                                                              *)
(*                                                                         *)
(* One single-threaded writer appends rows with ids 1, 2, 3, ...  A trim   *)
(* deletes rows with id <= latest - KEEP, so `oldest` (the smallest        *)
(* retained id) rises over time and oldest <= latest + 1 always.          *)
(*                                                                         *)
(* One client receives rows over a stream and tracks `cursor`, the id of  *)
(* the last row it received.  The stream can drop at any moment, and a    *)
(* drop can coincide with an append (the row is written to a dead socket  *)
(* and never arrives).  On reconnect the client presents its cursor and   *)
(* the server, in ONE atomic step:                                         *)
(*   1. if cursor < oldest - 1 the log no longer holds everything after    *)
(*      the cursor: it signals a GAP; the client records it and resyncs   *)
(*      at `latest`;                                                       *)
(*   2. otherwise it replays every retained row with id > cursor, in       *)
(*      order;                                                             *)
(*   3. and attaches the live listener, so every later append is          *)
(*      delivered while the stream stays open.                            *)
(* The step is atomic because the writer is single-threaded and the       *)
(* replay is synchronous with no yield before the listener is registered. *)
(* The mutant RoostPushSplitAttach splits it in two and fails.            *)
(***************************************************************************)

EXTENDS Naturals, FiniteSets

CONSTANTS
  MAX_ID,  \* appends stop here, keeping the model finite
  KEEP     \* retention: rows with id <= latest - KEEP may be trimmed

Ids == 1..MAX_ID

VARIABLES
  latest,    \* id of the last appended row (0 = empty log)
  oldest,    \* smallest retained id (latest + 1 when nothing is retained)
  open,      \* the stream is attached and live
  cursor,    \* client: id of the last row received (0 initially)
  received,  \* client: the set of ids it has received
  dupes,     \* client: how many deliveries repeated an id already received
  gapSeen,   \* client: the server signalled a gap at some resume
  lost,      \* history: a trim deleted a row the client had not received
  resumed    \* observation: the last step was a resume that signalled no gap

vars == <<latest, oldest, open, cursor, received, dupes, gapSeen, lost, resumed>>

TypeOK ==
  /\ latest \in 0..MAX_ID
  /\ oldest \in 1..(MAX_ID + 1)
  /\ oldest <= latest + 1
  /\ open \in BOOLEAN
  /\ cursor \in 0..MAX_ID
  /\ received \subseteq Ids
  /\ dupes \in Nat
  /\ gapSeen \in BOOLEAN
  /\ lost \in BOOLEAN
  /\ resumed \in BOOLEAN

Init ==
  /\ latest = 0
  /\ oldest = 1
  /\ open = FALSE
  /\ cursor = 0
  /\ received = {}
  /\ dupes = 0
  /\ gapSeen = FALSE
  /\ lost = FALSE
  /\ resumed = FALSE

(* ---------------------------------------------------------------- helpers *)

Deliver(S, ids) == S \cup ids

Repeats(S, ids) == Cardinality(S \cap ids)

(* ------------------------------------------------------------------ steps *)

(* The writer appends row latest + 1.  With the stream open the row is
   delivered live, unless the stream drops in the same instant: then the
   row went to a dead socket and the client never sees it. *)
Append ==
  /\ latest < MAX_ID
  /\ latest' = latest + 1
  /\ \/ /\ open
        /\ received' = Deliver(received, {latest'})
        /\ dupes' = dupes + Repeats(received, {latest'})
        /\ cursor' = latest'
        /\ UNCHANGED open
     \/ /\ open
        /\ open' = FALSE
        /\ UNCHANGED <<received, dupes, cursor>>
     \/ /\ ~open
        /\ UNCHANGED <<open, received, dupes, cursor>>
  /\ resumed' = FALSE
  /\ UNCHANGED <<oldest, gapSeen, lost>>

(* Retention: any prefix of the trimmable rows may go.  `lost` records that
   a row the client had not yet received was deleted. *)
Trim ==
  /\ \E o \in (oldest + 1)..(latest - KEEP + 1) :
       /\ oldest' = o
       /\ lost' = (lost \/ o > cursor + 1)
  /\ resumed' = FALSE
  /\ UNCHANGED <<latest, open, cursor, received, dupes, gapSeen>>

(* The stream drops with nothing in flight. *)
Drop ==
  /\ open
  /\ open' = FALSE
  /\ resumed' = FALSE
  /\ UNCHANGED <<latest, oldest, cursor, received, dupes, gapSeen, lost>>

(* Reconnect with Last-Event-ID = cursor: gap check, replay and attach in
   one atomic step. *)
Resume ==
  /\ ~open
  /\ open' = TRUE
  /\ IF cursor < oldest - 1
       THEN /\ gapSeen' = TRUE
            /\ cursor' = latest
            /\ resumed' = FALSE
            /\ UNCHANGED <<received, dupes>>
       ELSE LET replay == (cursor + 1)..latest IN
            /\ received' = Deliver(received, replay)
            /\ dupes' = dupes + Repeats(received, replay)
            /\ cursor' = latest
            /\ resumed' = TRUE
            /\ UNCHANGED gapSeen
  /\ UNCHANGED <<latest, oldest, lost>>

Next == Append \/ Trim \/ Drop \/ Resume

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

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

(* Unless the server told the client about a gap, the client holds exactly
   the contiguous prefix 1..cursor: nothing skipped. *)
NoSilentGap ==
  ~gapSeen => received = 1..cursor

(* No row is ever delivered twice. *)
NoDuplicates ==
  dupes = 0

(* Right after a resume that signalled no gap the client is current: the
   step is atomic, so no append can be in flight between replay and attach. *)
CaughtUpAfterResume ==
  resumed => cursor = latest

(* A gap is signalled only when retention really deleted a row the client
   had not received; a client whose cursor is still retained never sees one. *)
GapOnlyWhenTrimmed ==
  gapSeen => lost

(* While the stream is open the client is at the head of the log. *)
OpenStreamIsCurrent ==
  open => cursor = latest

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