session: name the two ways a durable wait ends without an acknowledgment

Review follow-up on the checkpoint store.

A dropped reply channel and an expired caller budget were both reported as
ReplyLost. They are different facts -- the first means the write is over and its
outcome did not reach here, the second means the save is still going -- so they are
now separate outcomes, and the durable wait has its own budget rather than borrowing
the one that bounds a call to a participant. Both still leave durable metadata where
it was, and for both the resolution asks the store about the same checkpoint.

The writer's two bounds refuse at different moments and the comment claimed
otherwise: the outstanding-capture bound is taken before a capture is requested, and
the byte budget cannot be, because a capture's size is not known until it exists. The
byte check, the decision and the change to the byte total are now one critical
section, the peak is sampled after a superseded job's bytes are gone, and the writer's
own bookkeeping is over a type that holds only the outcomes a writer can produce.

The manifest's coordinator.eventWatermarks is {lastSourceStep, issued}; the fixture
illustrated {lastEventId, lastOrdinal}, and the illustration is what changed, because
an event id is derived from the epoch and cannot be compared across the restore that
gives the session a new one.

checkpoint-envelope-v1 section 3 also now says, under the same dated amendment, that
a required-manifest-field change must bump envelopeVersion once production files
exist: contractDigest is taken over the schema set and does not cover this manifest,
so the envelope version is the only thing that can carry such a change.

docs/design/session-framework/checkpoint-envelope-v1.md

@@ -118,6 +118,18 @@ section has listed from the start. Both are now in `REQUIRED_MANIFEST_FIELDS` in
 TypeScript, and the fixture was regenerated by the existing example. The schema set is
 untouched, so `contractDigest` is unchanged.
 
+`coordinator.eventWatermarks` is `{lastSourceStep, issued}`. The fixture illustrated
+`{lastEventId, lastOrdinal}`, and it is the illustration that changed: an event id is derived
+from the epoch, so a watermark spelled as one cannot be compared across the restore that
+gives the session a new epoch, while a source step and an issued count can.
+
+**A required-manifest-field change is compatibility-relevant and `contractDigest` does not
+cover it.** The digest is taken over the schema set, and this manifest is not in it, so
+`envelopeVersion` is the only thing that can carry such a change. It stays `1` here only
+because no production `FLYSESS1` file exists yet: once one does, adding or removing a required
+manifest field **must** bump `envelopeVersion`, because a reader of the older version would
+otherwise accept a file it cannot completely read, or refuse one it could.
+
 `payloads` is redundant with the table on purpose: the table is what a reader needs to map
 bytes, and the manifest is what a store lists, compares and reports without opening the
 payload area. A reader checks that the two agree.

services/flysim/crates/fly-session-types/examples/update_fixtures.rs

@@ -197,7 +197,7 @@ fn checkpoint_envelope() -> String {
             "priorInspection": "prior-inspection",
             "executorState": [{"agentId": "fly-a", "payload": "executor-fly-a"}],
             "admissionState": null,
-            "eventWatermarks": {"lastEventId": "evt-1", "lastOrdinal": "7"},
+            "eventWatermarks": {"lastSourceStep": "42", "issued": "7"},
         },
         "environment": {"workerId": "arena", "payload": "world"},
         "helperState": [],

services/flysim/crates/fly-session-types/fixtures/checkpoint-envelope.json

@@ -59,8 +59,8 @@
       ],
       "admissionState": null,
       "eventWatermarks": {
-        "lastEventId": "evt-1",
+        "lastSourceStep": "42",
-        "lastOrdinal": "7"
+        "issued": "7"
       }
     },
     "environment": {
@@ -119,49 +119,49 @@
     }
   ],
   "envelope": {
-    "base64": "RkxZU0VTUzEBAAAAIAAAADUIAAAFAAAAWAgAAAAAAAB7ImFnZW50cyI6W3siYWdlbnRJZCI6ImZseS1hIiwiYnJhaW5UaWNrcyI6IjI1MzQiLCJkYXRhc2V0RGlnZXN0IjoiNmMwYWYxZjA3ODRlZjYzYTM5M2VlNzdkNjE0ZTgyNDZjNjI1MDUxMzYwZjNmMWE0ODgzODM3NGM1ZDM1NWI1MiIsIm1vZGVsVmVyc2lvbiI6ImxpZi0xbXMtZjY0LXYyIiwicGF5bG9hZCI6ImFnZW50LWZseS1hIiwicGxhc3RpY2l0eVZlcnNpb24iOiJmbHkta2MtbWJvbi1yc3RkcC12MiIsInByb2ZpbGVEaWdlc3QiOiIxOTAwZWFiNmMwMjg0ODNkNzEyNjU5OWVlNmY1MGRlMGQyNzkwN2I1YzY1ZmE5MDUyNDU4MGI0YjBmOTg1MmIwIiwicmVtYWluZGVyIjp7ImRlbm9taW5hdG9yIjoiMyIsIm51bWVyYXRvciI6IjEwMDAwMDAifSwic2VlZCI6LTE4NDk0NjA2M31dLCJjaGVja3BvaW50SWQiOiJja3B0LTEiLCJjb21wYXRpYmlsaXR5Ijp7ImJhY2tlbmREaWdlc3QiOiIxMGUwOGE0MTllODUwZWJhMWViYmExOGZkZDI4ZWI3ZWMxYjdlOGJhYTliY2MzYjk3M2UyYjg4OTFlYzcyNmJlIiwiY29udGVudERpZ2VzdCI6ImVkNzAwMmI0MzllOWFjODQ1ZjIyMzU3ZDgyMmJhYzE0NDQ3MzBmYmRiNjAxNmQzZWM5NDMyMjk3YjllYzlmNzMiLCJjb250cm9sbGVyRGlnZXN0IjoiYzE0NzIxMzViMTRjNzdjOGJlZjk4ZTczZjcwMjA4MzI1ZmEwZGNmMWU2YmQ2NjhhZTliMzFhOWNlYTI5NWZlNyIsInBhcnNlckRpZ2VzdCI6ImIxN2Q0NTEyMTE1MDkyOGYyMTQ2YWY0OWUxOTVlZmYxZWVmNWQ2NzMyNWJlMjczYTczM2ZiNzRhY2FkYWEzNDIiLCJwYXRjaERpZ2VzdCI6ImE0ODk1ZWI0NGFmYzMzNmZlY2JiYTZlNTIwY2Q2N2UxNzhkYWNlMDI3NjY1NWQxMDJmY2VmZmE4ZTVmNzA1NzAiLCJzdGF0ZUZvcm1hdElkIjoiZmx5c2Vzcy0xIn0sImNvbXBvc2l0aW9uRGlnZXN0IjoiNzMwZDcyNWM4YTU5ZDNhNzMwM2RlZjJiZWQwNDFhNTc3ZWRiNDI1NWFhYmQ0ODg5Y2UxMjkxODMxMWQ5NTJmMCIsImNvb3JkaW5hdG9yIjp7ImFkbWlzc2lvblN0YXRlIjpudWxsLCJldmVudFdhdGVybWFya3MiOnsibGFzdEV2ZW50SWQiOiJldnQtMSIsImxhc3RPcmRpbmFsIjoiNyJ9LCJleGVjdXRvclN0YXRlIjpbeyJhZ2VudElkIjoiZmx5LWEiLCJwYXlsb2FkIjoiZXhlY3V0b3ItZmx5LWEifV0sInByaW9ySW5zcGVjdGlvbiI6InByaW9yLWluc3BlY3Rpb24iLCJ0YXNrTGVkZ2VyIjoidGFzay1sZWRnZXIifSwiZW52ZWxvcGVWZXJzaW9uIjoxLCJlbnZpcm9ubWVudCI6eyJwYXlsb2FkIjoid29ybGQiLCJ3b3JrZXJJZCI6ImFyZW5hIn0sImVwaXNvZGVJZCI6ImVwaXNvZGUtMSIsImhlbHBlclN0YXRlIjpbXSwicGF5bG9hZHMiOlt7ImJ5dGVMZW5ndGgiOiIxNyIsImRpZ2VzdCI6IjEzMjFkZmZiMGNkYzZmOTA5MmNiZjdmYTJhNWZjNjhiYmVkMTJjOTkzZDVhZDM5ODI2NDAxMjgxMGNlOWJmOTMiLCJuYW1lIjoiYWdlbnQtZmx5LWEifSx7ImJ5dGVMZW5ndGgiOiIxNCIsImRpZ2VzdCI6IjNhZWU2MGRmN2UyOWVmZWJhN2Y1Zjk5ZmM1ODY3NjQ3YjM2YWViZmYxZDVkM2M4MzhkYmZmMzIzMTIyZTY0NjIiLCJuYW1lIjoiZXhlY3V0b3ItZmx5LWEifSx7ImJ5dGVMZW5ndGgiOiIxMSIsImRpZ2VzdCI6IjQwYjAwZWQyYmJiYTkwMWQ2ODIwNWZmNzFiMDRhNDRiOWVlNTNjNTFjYjMxMDlhYTJjZWFhNDRmMWM0NTcyN2UiLCJuYW1lIjoidGFzay1sZWRnZXIifSx7ImJ5dGVMZW5ndGgiOiIxMCIsImRpZ2VzdCI6IjJjMTNiN2I0ZDlhOTkxNjgwMWFiOTE5MWMzMTRmMzFiMDQ1ZTliOWM1YjY2OWE2YzA0NzRmMDIxN2VmNzViZjUiLCJuYW1lIjoicHJpb3ItaW5zcGVjdGlvbiJ9LHsiYnl0ZUxlbmd0aCI6IjY0IiwiZGlnZXN0IjoiZjVhNWZkNDJkMTZhMjAzMDI3OThlZjZlZDMwOTk3OWI0MzAwM2QyMzIwZDlmMGU4ZWE5ODMxYTkyNzU5ZmI0YiIsIm5hbWUiOiJ3b3JsZCJ9XSwicG9ydE1hcCI6W3siYWdlbnRJZCI6ImZseS1hIiwicG9ydElkIjoicG9ydC0xIn1dLCJzY2hlZHVsZXJJZCI6ImxvY2tzdGVwLXYxIiwic291cmNlU2NvcGUiOnsiZXBvY2giOiJlcG9jaC0xIiwic2Vzc2lvbklkIjoiZGVtbyIsInN0ZXAiOiI0MiJ9LCJ3b3JsZFRpbWUiOnsiZGVub21pbmF0b3IiOiIxIiwibnVtZXJhdG9yIjoiNzAwMDAwMDAwIn19AAAAYWdlbnQtZmx5LWEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAIgKAAAAAAAAEQAAAAAAAAATId/7DNxvkJLL9/oqX8aLvtEsmT1a05gmQBKBDOm/k2V4ZWN1dG9yLWZseS1hAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAACgCgAAAAAAAA4AAAAAAAAAOu5g334p7+un9fmfxYZ2R7Nq6/8dXTyDjb/zIxIuZGJ0YXNrLWxlZGdlcgAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAsAoAAAAAAAALAAAAAAAAAECwDtK7upAdaCBf9xsEpEue5TxRyzEJqizqpE8cRXJ+cHJpb3ItaW5zcGVjdGlvbgAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAMAKAAAAAAAACgAAAAAAAAAsE7e02amRaAGrkZHDFPMbBF6bnFtmmmwEdPAhfvdb9XdvcmxkAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAADQCgAAAAAAAEAAAAAAAAAA9aX9QtFqIDAnmO9u0wmXm0MAPSMg2fDo6pgxqSdZ+0thZ2VudCBzdGF0ZSBieXRlcwAAAAAAAABleGVjdXRvciBzdGF0ZQAAeyJyYW5rIjoxMH0AAAAAAHsibWFwIjo0MH0AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAQAsAAAAAAAAhBlh9AWLTtVKckmeIzNn4DO5Yhn2C1nUA1T1RfxMXlkZMWVNFU1NG",
+    "base64": "RkxZU0VTUzEBAAAAIAAAADAIAAAFAAAAUAgAAAAAAAB7ImFnZW50cyI6W3siYWdlbnRJZCI6ImZseS1hIiwiYnJhaW5UaWNrcyI6IjI1MzQiLCJkYXRhc2V0RGlnZXN0IjoiNmMwYWYxZjA3ODRlZjYzYTM5M2VlNzdkNjE0ZTgyNDZjNjI1MDUxMzYwZjNmMWE0ODgzODM3NGM1ZDM1NWI1MiIsIm1vZGVsVmVyc2lvbiI6ImxpZi0xbXMtZjY0LXYyIiwicGF5bG9hZCI6ImFnZW50LWZseS1hIiwicGxhc3RpY2l0eVZlcnNpb24iOiJmbHkta2MtbWJvbi1yc3RkcC12MiIsInByb2ZpbGVEaWdlc3QiOiIxOTAwZWFiNmMwMjg0ODNkNzEyNjU5OWVlNmY1MGRlMGQyNzkwN2I1YzY1ZmE5MDUyNDU4MGI0YjBmOTg1MmIwIiwicmVtYWluZGVyIjp7ImRlbm9taW5hdG9yIjoiMyIsIm51bWVyYXRvciI6IjEwMDAwMDAifSwic2VlZCI6LTE4NDk0NjA2M31dLCJjaGVja3BvaW50SWQiOiJja3B0LTEiLCJjb21wYXRpYmlsaXR5Ijp7ImJhY2tlbmREaWdlc3QiOiIxMGUwOGE0MTllODUwZWJhMWViYmExOGZkZDI4ZWI3ZWMxYjdlOGJhYTliY2MzYjk3M2UyYjg4OTFlYzcyNmJlIiwiY29udGVudERpZ2VzdCI6ImVkNzAwMmI0MzllOWFjODQ1ZjIyMzU3ZDgyMmJhYzE0NDQ3MzBmYmRiNjAxNmQzZWM5NDMyMjk3YjllYzlmNzMiLCJjb250cm9sbGVyRGlnZXN0IjoiYzE0NzIxMzViMTRjNzdjOGJlZjk4ZTczZjcwMjA4MzI1ZmEwZGNmMWU2YmQ2NjhhZTliMzFhOWNlYTI5NWZlNyIsInBhcnNlckRpZ2VzdCI6ImIxN2Q0NTEyMTE1MDkyOGYyMTQ2YWY0OWUxOTVlZmYxZWVmNWQ2NzMyNWJlMjczYTczM2ZiNzRhY2FkYWEzNDIiLCJwYXRjaERpZ2VzdCI6ImE0ODk1ZWI0NGFmYzMzNmZlY2JiYTZlNTIwY2Q2N2UxNzhkYWNlMDI3NjY1NWQxMDJmY2VmZmE4ZTVmNzA1NzAiLCJzdGF0ZUZvcm1hdElkIjoiZmx5c2Vzcy0xIn0sImNvbXBvc2l0aW9uRGlnZXN0IjoiNzMwZDcyNWM4YTU5ZDNhNzMwM2RlZjJiZWQwNDFhNTc3ZWRiNDI1NWFhYmQ0ODg5Y2UxMjkxODMxMWQ5NTJmMCIsImNvb3JkaW5hdG9yIjp7ImFkbWlzc2lvblN0YXRlIjpudWxsLCJldmVudFdhdGVybWFya3MiOnsiaXNzdWVkIjoiNyIsImxhc3RTb3VyY2VTdGVwIjoiNDIifSwiZXhlY3V0b3JTdGF0ZSI6W3siYWdlbnRJZCI6ImZseS1hIiwicGF5bG9hZCI6ImV4ZWN1dG9yLWZseS1hIn1dLCJwcmlvckluc3BlY3Rpb24iOiJwcmlvci1pbnNwZWN0aW9uIiwidGFza0xlZGdlciI6InRhc2stbGVkZ2VyIn0sImVudmVsb3BlVmVyc2lvbiI6MSwiZW52aXJvbm1lbnQiOnsicGF5bG9hZCI6IndvcmxkIiwid29ya2VySWQiOiJhcmVuYSJ9LCJlcGlzb2RlSWQiOiJlcGlzb2RlLTEiLCJoZWxwZXJTdGF0ZSI6W10sInBheWxvYWRzIjpbeyJieXRlTGVuZ3RoIjoiMTciLCJkaWdlc3QiOiIxMzIxZGZmYjBjZGM2ZjkwOTJjYmY3ZmEyYTVmYzY4YmJlZDEyYzk5M2Q1YWQzOTgyNjQwMTI4MTBjZTliZjkzIiwibmFtZSI6ImFnZW50LWZseS1hIn0seyJieXRlTGVuZ3RoIjoiMTQiLCJkaWdlc3QiOiIzYWVlNjBkZjdlMjllZmViYTdmNWY5OWZjNTg2NzY0N2IzNmFlYmZmMWQ1ZDNjODM4ZGJmZjMyMzEyMmU2NDYyIiwibmFtZSI6ImV4ZWN1dG9yLWZseS1hIn0seyJieXRlTGVuZ3RoIjoiMTEiLCJkaWdlc3QiOiI0MGIwMGVkMmJiYmE5MDFkNjgyMDVmZjcxYjA0YTQ0YjllZTUzYzUxY2IzMTA5YWEyY2VhYTQ0ZjFjNDU3MjdlIiwibmFtZSI6InRhc2stbGVkZ2VyIn0seyJieXRlTGVuZ3RoIjoiMTAiLCJkaWdlc3QiOiIyYzEzYjdiNGQ5YTk5MTY4MDFhYjkxOTFjMzE0ZjMxYjA0NWU5YjljNWI2NjlhNmMwNDc0ZjAyMTdlZjc1YmY1IiwibmFtZSI6InByaW9yLWluc3BlY3Rpb24ifSx7ImJ5dGVMZW5ndGgiOiI2NCIsImRpZ2VzdCI6ImY1YTVmZDQyZDE2YTIwMzAyNzk4ZWY2ZWQzMDk5NzliNDMwMDNkMjMyMGQ5ZjBlOGVhOTgzMWE5Mjc1OWZiNGIiLCJuYW1lIjoid29ybGQifV0sInBvcnRNYXAiOlt7ImFnZW50SWQiOiJmbHktYSIsInBvcnRJZCI6InBvcnQtMSJ9XSwic2NoZWR1bGVySWQiOiJsb2Nrc3RlcC12MSIsInNvdXJjZVNjb3BlIjp7ImVwb2NoIjoiZXBvY2gtMSIsInNlc3Npb25JZCI6ImRlbW8iLCJzdGVwIjoiNDIifSwid29ybGRUaW1lIjp7ImRlbm9taW5hdG9yIjoiMSIsIm51bWVyYXRvciI6IjcwMDAwMDAwMCJ9fWFnZW50LWZseS1hAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAACACgAAAAAAABEAAAAAAAAAEyHf+wzcb5CSy/f6Kl/Gi77RLJk9WtOYJkASgQzpv5NleGVjdXRvci1mbHktYQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAmAoAAAAAAAAOAAAAAAAAADruYN9+Ke/rp/X5n8WGdkezauv/HV08g42/8yMSLmRidGFzay1sZWRnZXIAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAKgKAAAAAAAACwAAAAAAAABAsA7Su7qQHWggX/cbBKRLnuU8UcsxCaos6qRPHEVyfnByaW9yLWluc3BlY3Rpb24AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAC4CgAAAAAAAAoAAAAAAAAALBO3tNmpkWgBq5GRwxTzGwRem5xbZppsBHTwIX73W/V3b3JsZAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAyAoAAAAAAABAAAAAAAAAAPWl/ULRaiAwJ5jvbtMJl5tDAD0jINnw6OqYMaknWftLYWdlbnQgc3RhdGUgYnl0ZXMAAAAAAAAAZXhlY3V0b3Igc3RhdGUAAHsicmFuayI6MTB9AAAAAAB7Im1hcCI6NDB9AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAADgLAAAAAAAAX4L9WkdX0MViD3h5YJgf7VQgwocWU4XJoKX6MUwl+8hGTFlTRVNTRg==",
-    "byteLength": 2880,
+    "byteLength": 2872,
     "layout": {
       "headerBytes": 32,
       "manifestOffset": "32",
-      "manifestBytes": 2101,
+      "manifestBytes": 2096,
-      "tableOffset": "2136",
+      "tableOffset": "2128",
       "tableEntryBytes": 112,
       "entries": [
         {
           "name": "agent-fly-a",
-          "offset": "2696",
+          "offset": "2688",
           "byteLength": "17",
           "digest": "1321dffb0cdc6f9092cbf7fa2a5fc68bbed12c993d5ad398264012810ce9bf93"
         },
         {
           "name": "executor-fly-a",
-          "offset": "2720",
+          "offset": "2712",
           "byteLength": "14",
           "digest": "3aee60df7e29efeba7f5f99fc5867647b36aebff1d5d3c838dbff323122e6462"
         },
         {
           "name": "task-ledger",
-          "offset": "2736",
+          "offset": "2728",
           "byteLength": "11",
           "digest": "40b00ed2bbba901d68205ff71b04a44b9ee53c51cb3109aa2ceaa44f1c45727e"
         },
         {
           "name": "prior-inspection",
-          "offset": "2752",
+          "offset": "2744",
           "byteLength": "10",
           "digest": "2c13b7b4d9a9916801ab9191c314f31b045e9b9c5b669a6c0474f0217ef75bf5"
         },
         {
           "name": "world",
-          "offset": "2768",
+          "offset": "2760",
           "byteLength": "64",
           "digest": "f5a5fd42d16a20302798ef6ed309979b43003d2320d9f0e8ea9831a92759fb4b"
         }
       ],
-      "footerOffset": "2832",
+      "footerOffset": "2824",
       "footerBytes": 48,
-      "totalBytes": "2880"
+      "totalBytes": "2872"
     }
   },
   "corruption": [
@@ -182,17 +182,17 @@
     },
     {
       "name": "a flipped payload byte",
-      "offset": 2696,
+      "offset": 2688,
       "reason": "every payload carries its own digest"
     },
     {
       "name": "a flipped footer digest byte",
-      "offset": 2840,
+      "offset": 2832,
       "reason": "the footer digest must match the contents"
     },
     {
       "name": "a flipped footer magic byte",
-      "offset": 2872,
+      "offset": 2864,
       "reason": "a truncated file cannot look complete"
     }
   ]

services/flysim/crates/fly-session/README.md

@@ -189,12 +189,18 @@ The durable store is `state`, over the `FLYSESS1` layout the contract crate owns
   `BUSY` a stepping session survives rather than an epoch failure.
 - **Capture and durability are two events.** `State.Capture` completes when an immutable
   capture exists; `Coordinator::await_durable` completes when the store manifest rename has
-  happened, which is the durable commit point. Only the second moves the durable mark. A lost
+  happened, which is the durable commit point. Only the second moves the durable mark, and the
-  save reply is `SaveOutcome::ReplyLost`, and `Coordinator::resolve_durable` then asks the
+  three ways it can end without one are told apart: `Failed` (the write stopped),
-  store about the *same* checkpoint instead of saving again.
+  `ReplyLost` (the write finished and the acknowledgment did not arrive) and
-- **The writer is bounded twice**, by outstanding captures and by queued bytes, and it owns
+  `DeadlineExpired` (the caller's own budget ran out while the save was still going).
-  its payload handles until the bytes are committed or the job fails. A queued *replaceable*
+  `Coordinator::resolve_durable` then asks the store about the *same* checkpoint instead of
-  capture is superseded by a later one, releasing its holds; a durable one never is.
+  saving again.
+- **The writer is bounded twice**, and the two bounds refuse at different moments. The
+  outstanding-capture bound is taken before a capture is requested; the byte budget cannot be,
+  because a capture's size is not known until it exists, so it refuses at submit and releases
+  the payloads with the refusal. The writer owns its payload handles until the bytes are
+  committed or the job fails. A queued *replaceable* capture is superseded by a later one,
+  releasing its holds; a durable one never is.
 - **The install is a group.** A restore selects a complete compatible generation, imports its
   payloads as fresh artifacts, stages every participant, validates the coordinator's own
   ledgers, and only then activates. A failure anywhere leaves the fence closed, and every

services/flysim/crates/fly-session/src/coordinator.rs

@@ -145,6 +145,12 @@ pub struct Deadlines {
     /// a capture serializes a participant and a restore validates and installs one, and
     /// neither is a step whose latency the probe was chosen for.
     pub capture: Duration,
+    /// How long a caller waits for a *durable* acknowledgment.
+    ///
+    /// It is not [`Deadlines::capture`]: that one bounds a call to a participant, and this
+    /// one bounds two `fsync`s, a queue the caller shares with other captures and a disk.
+    /// Reusing the call budget here would make a slow disk look like an unresponsive worker.
+    pub durable: Duration,
 }
 
 /// How long the resolution waits between attempts.
@@ -172,6 +178,7 @@ impl Default for Deadlines {
             resolve_attempts: 8192,
             boot: Duration::from_secs(30),
             capture: Duration::from_secs(30),
+            durable: Duration::from_secs(60),
         }
     }
 }
@@ -3186,14 +3193,17 @@ impl Coordinator {
     /// operation.
     pub async fn await_durable(&mut self, ticket: CaptureTicket) -> Outcome<crate::state::SaveOutcome> {
         let CaptureTicket { checkpoint_id, boundary, receiver } = ticket;
-        let budget = self.deadlines.capture;
+        let budget = self.deadlines.durable;
         let outcome =
             crate::state::CheckpointWriter::wait(receiver, &checkpoint_id, budget).await;
-        if let crate::state::SaveOutcome::Committed { .. } = &outcome {
+        if outcome.is_durable() {
             self.durable = Some((checkpoint_id.clone(), boundary));
             self.audit.push(format!("durable:{checkpoint_id}@{boundary}"));
         } else {
-            self.audit.push(format!("not-durable:{checkpoint_id}@{boundary}"));
+            // Named rather than lumped together: a failed write, a superseded capture, a lost
+            // reply and an expired caller budget are four different things to have to explain.
+            self.audit
+                .push(format!("not-durable:{}:{checkpoint_id}@{boundary}", outcome.event()));
         }
         Ok(outcome)
     }

services/flysim/crates/fly-session/src/state.rs

@@ -999,10 +999,18 @@ pub enum SaveOutcome {
     Failed { checkpoint_id: Id, reason: String },
     /// A later replaceable capture took this one's place in the queue before it was written.
     Superseded { checkpoint_id: Id, by: Id },
-    /// The writer's reply never arrived. The operation's outcome is unknown from here, so
+    /// The writer finished this job and its reply channel was gone before the outcome could
-    /// durable metadata does not move; the caller resolves the *same* operation against the
+    /// be delivered. The write is over and its result is unknown from here.
-    /// store manifest instead of saving again.
     ReplyLost { checkpoint_id: Id },
+    /// The caller's own budget ran out while the job was still queued or being written. The
+    /// save is not over: it may commit after this is reported.
+    ///
+    /// It is a different fact from [`SaveOutcome::ReplyLost`] and is named separately because
+    /// diagnosing one as the other is exactly the implicit best-effort reading these
+    /// contracts refuse. Both leave durable metadata where it was, and for both the caller
+    /// resolves the *same* operation against the store manifest instead of saving again --
+    /// but only one of them is a save that has already stopped.
+    DeadlineExpired { checkpoint_id: Id },
 }
 
 impl SaveOutcome {
@@ -1011,17 +1019,54 @@ impl SaveOutcome {
             SaveOutcome::Committed { checkpoint_id, .. }
             | SaveOutcome::Failed { checkpoint_id, .. }
             | SaveOutcome::Superseded { checkpoint_id, .. }
-            | SaveOutcome::ReplyLost { checkpoint_id } => checkpoint_id,
+            | SaveOutcome::ReplyLost { checkpoint_id }
+            | SaveOutcome::DeadlineExpired { checkpoint_id } => checkpoint_id,
         }
     }
 
     /// The event name this outcome publishes under.
+    ///
+    /// Only the three the writer itself produces are ever published; the two caller-side
+    /// outcomes are what a caller saw, not what the store did, and the store does not announce
+    /// them on its own topic.
     pub fn event(&self) -> &'static str {
         match self {
             SaveOutcome::Committed { .. } => "committed",
             SaveOutcome::Failed { .. } => "failed",
             SaveOutcome::Superseded { .. } => "superseded",
-            SaveOutcome::ReplyLost { .. } => "failed",
+            SaveOutcome::ReplyLost { .. } | SaveOutcome::DeadlineExpired { .. } => "failed",
+        }
+    }
+
+    /// True only past the durable commit point.
+    pub fn is_durable(&self) -> bool {
+        matches!(self, SaveOutcome::Committed { .. })
+    }
+}
+
+/// What the writer itself can produce for one job.
+///
+/// The two caller-side outcomes -- a lost reply and an expired caller deadline -- are not in
+/// here, because the writer cannot observe either. Keeping them out is what stops the writer's
+/// own bookkeeping from carrying arms that can never run.
+#[derive(Clone, Debug, PartialEq, Eq)]
+enum WriteOutcome {
+    Committed { boundary: u64, file: String },
+    Failed { reason: String },
+}
+
+impl WriteOutcome {
+    fn into_save(self, checkpoint_id: &Id) -> SaveOutcome {
+        match self {
+            WriteOutcome::Committed { boundary, file } => SaveOutcome::Committed {
+                checkpoint_id: checkpoint_id.clone(),
+                boundary,
+                file,
+            },
+            WriteOutcome::Failed { reason } => SaveOutcome::Failed {
+                checkpoint_id: checkpoint_id.clone(),
+                reason,
+            },
         }
     }
 }
@@ -1037,13 +1082,21 @@ pub enum RetryPolicy {
     RetryThenRelease { attempts: u32 },
 }
 
-/// The writer's bounds. Both are finite and both refuse before a capture is requested.
+/// The writer's bounds. Both are finite, and they refuse at different moments.
+///
+/// [`WriterConfig::queue_capacity`] is the one a capture is refused *before* it is requested:
+/// [`CheckpointWriter::reserve`] takes its slot first, which is what the durable row of
+/// `state-media-v1` section 3 means by rejecting before capture. The byte budget cannot work
+/// that way, because how many bytes a capture is worth is not known until the participants
+/// have produced it; it is checked at [`CheckpointWriter::submit`], so an oversized capture is
+/// refused after it exists and before it is queued, and its payloads are released with the
+/// refusal. Both are named `BUSY` refusals and neither fails the epoch.
 #[derive(Clone, Copy, Debug)]
 pub struct WriterConfig {
     /// Outstanding coherent captures. `state-media-v1` section 3's initial session default
-    /// is two.
+    /// is two. Refused before a capture is requested.
     pub queue_capacity: usize,
-    /// The total payload bytes the queue may hold.
+    /// The total payload bytes the queue may hold. Refused at submit, once the size is known.
     pub max_queued_bytes: u64,
     pub retry: RetryPolicy,
 }
@@ -1198,48 +1251,57 @@ capture is refused before it is requested rather than queued without bound",
     ) -> DomainResult<tokio::sync::oneshot::Receiver<SaveOutcome>> {
         let bytes = submission.byte_length();
         let budget = self.shared.config.max_queued_bytes;
-        let held = self.shared.queued_bytes.load(Ordering::SeqCst);
-        if held + bytes > budget {
-            self.shared
-                .stats
-                .lock()
-                .expect("the writer stats are never poisoned")
-                .rejected += 1;
-            return Err(DomainError::before(
-                ErrorCode::Busy,
-                format!(
-                    "the checkpoint queue holds {held} of {budget} bytes and this capture adds \
-{bytes}; the byte budget is finite and refuses before it is exceeded"
-                ),
-            ));
-        }
         let (reply, receiver) = tokio::sync::oneshot::channel();
         let checkpoint_id = submission.checkpoint_id.clone();
         let queued_event = submission.as_event("queued");
+        // Reading the byte total, deciding on it and changing it are one critical section.
+        // Only one caller submits today, so a split could not be observed -- but a bound that
+        // is only correct while nobody else is submitting is not a bound.
         let superseded = {
             let mut queue = self.shared.queue.lock().expect("the writer queue is never poisoned");
-            let replaced = if submission.replaceable {
+            let replaced_index = if submission.replaceable {
-                queue
+                queue.iter().position(|job| job.submission.replaceable)
-                    .iter()
-                    .position(|job| job.submission.replaceable)
-                    .map(|index| queue.remove(index).expect("just found"))
             } else {
                 None
             };
+            // A capture that will take a queued replaceable one's place frees its bytes, so
+            // the budget is decided against what the queue will hold and not what it holds.
+            let freed = replaced_index.map_or(0, |index| queue[index].submission.byte_length());
+            let held = self.shared.queued_bytes.load(Ordering::SeqCst);
+            let after = held.saturating_sub(freed) + bytes;
+            if after > budget {
+                self.shared
+                    .stats
+                    .lock()
+                    .expect("the writer stats are never poisoned")
+                    .rejected += 1;
+                return Err(DomainError::before(
+                    ErrorCode::Busy,
+                    format!(
+                        "the checkpoint queue would hold {after} of {budget} bytes; the byte \
+budget is finite and refuses before it is exceeded"
+                    ),
+                ));
+            }
+            let replaced = replaced_index.map(|index| queue.remove(index).expect("just found"));
+            if let Some(old) = &replaced {
+                self.shared
+                    .queued_bytes
+                    .fetch_sub(old.submission.byte_length(), Ordering::SeqCst);
+            }
             queue.push_back(Job { submission, reply, permit: reservation.permit });
             self.shared.queued_bytes.fetch_add(bytes, Ordering::SeqCst);
             let mut stats = self.shared.stats.lock().expect("the writer stats are never poisoned");
             stats.queued += 1;
             stats.peak_queue = stats.peak_queue.max(queue.len());
+            // Sampled after the superseded job's bytes are gone, so the peak is a total the
+            // queue really held.
             stats.peak_bytes = stats
                 .peak_bytes
                 .max(self.shared.queued_bytes.load(Ordering::SeqCst));
             replaced
         };
         if let Some(old) = superseded {
-            self.shared
-                .queued_bytes
-                .fetch_sub(old.submission.byte_length(), Ordering::SeqCst);
             self.shared
                 .stats
                 .lock()
@@ -1267,8 +1329,12 @@ capture is refused before it is requested rather than queued without bound",
         Ok(receiver)
     }
 
-    /// Waits for one save's outcome. A dropped reply channel is a lost save reply, which is
+    /// Waits for one save's outcome, within the caller's own budget.
-    /// an outcome and not a hang.
+    ///
+    /// The two ways this ends without an outcome are different facts and are reported as
+    /// themselves: the channel closing means the writer finished and the reply did not reach
+    /// here, and the budget running out means the save is still going. Neither is a hang and
+    /// neither is a save.
     pub async fn wait(
         receiver: tokio::sync::oneshot::Receiver<SaveOutcome>,
         checkpoint_id: &Id,
@@ -1276,7 +1342,8 @@ capture is refused before it is requested rather than queued without bound",
     ) -> SaveOutcome {
         match tokio::time::timeout(budget, receiver).await {
             Ok(Ok(outcome)) => outcome,
-            Ok(Err(_)) | Err(_) => SaveOutcome::ReplyLost { checkpoint_id: checkpoint_id.clone() },
+            Ok(Err(_)) => SaveOutcome::ReplyLost { checkpoint_id: checkpoint_id.clone() },
+            Err(_) => SaveOutcome::DeadlineExpired { checkpoint_id: checkpoint_id.clone() },
         }
     }
 
@@ -1342,6 +1409,10 @@ fn outcome_event(submission: &CaptureSubmission, outcome: &SaveOutcome) -> Map<S
             payload.insert("reason".into(), "the save reply was lost".into());
             payload.insert("durable".into(), false.into());
         }
+        SaveOutcome::DeadlineExpired { .. } => {
+            payload.insert("reason".into(), "the caller's durable budget expired".into());
+            payload.insert("durable".into(), false.into());
+        }
     }
     payload
 }
@@ -1388,15 +1459,15 @@ async fn run_writer(shared: Arc<WriterShared>) {
         shared
             .queued_bytes
             .fetch_sub(submission.byte_length(), Ordering::SeqCst);
-        let outcome = write_one(&shared, &submission).await;
+        let written = write_one(&shared, &submission).await;
         {
             let mut stats = shared.stats.lock().expect("the writer stats are never poisoned");
-            match &outcome {
+            match &written {
-                SaveOutcome::Committed { .. } => stats.committed += 1,
+                WriteOutcome::Committed { .. } => stats.committed += 1,
-                SaveOutcome::Failed { .. } | SaveOutcome::ReplyLost { .. } => stats.failed += 1,
+                WriteOutcome::Failed { .. } => stats.failed += 1,
-                SaveOutcome::Superseded { .. } => stats.superseded += 1,
             }
         }
+        let outcome = written.into_save(&submission.checkpoint_id);
         publish_event(&shared.events, &shared.stats, &outcome_event(&submission, &outcome)).await;
         let lost = shared.faults.drop_reply_for.as_ref() == Some(&submission.checkpoint_id);
         // The writer owned these handles until the bytes were committed or the job failed.
@@ -1416,21 +1487,19 @@ async fn run_writer(shared: Arc<WriterShared>) {
     }
 }
 
-async fn write_one(shared: &Arc<WriterShared>, submission: &CaptureSubmission) -> SaveOutcome {
+async fn write_one(shared: &Arc<WriterShared>, submission: &CaptureSubmission) -> WriteOutcome {
     let mut payloads = Vec::with_capacity(submission.payloads.len());
     for payload in &submission.payloads {
         let bytes = match payload.artifact.read_all().await {
             Ok(bytes) => bytes,
             Err(e) => {
-                return SaveOutcome::Failed {
+                return WriteOutcome::Failed {
-                    checkpoint_id: submission.checkpoint_id.clone(),
                     reason: format!("payload {}: {}", payload.name, e.message),
                 };
             }
         };
         if bytes.len() as u64 != payload.byte_length || digest_of_bytes(&bytes) != payload.digest {
-            return SaveOutcome::Failed {
+            return WriteOutcome::Failed {
-                checkpoint_id: submission.checkpoint_id.clone(),
                 reason: format!(
                     "payload {} is not the content its capture declared",
                     payload.name
@@ -1442,8 +1511,7 @@ async fn write_one(shared: &Arc<WriterShared>, submission: &CaptureSubmission) -
     let bytes = match checkpoint::encode(&submission.manifest, &payloads) {
         Ok(bytes) => bytes,
         Err(e) => {
-            return SaveOutcome::Failed {
+            return WriteOutcome::Failed {
-                checkpoint_id: submission.checkpoint_id.clone(),
                 reason: format!("envelope: {}", e.0),
             };
         }
@@ -1477,8 +1545,7 @@ async fn write_one(shared: &Arc<WriterShared>, submission: &CaptureSubmission) -
         .await;
         match result {
             Ok(Ok(())) => {
-                return SaveOutcome::Committed {
+                return WriteOutcome::Committed {
-                    checkpoint_id: submission.checkpoint_id.clone(),
                     boundary: submission.boundary,
                     file: format!("{}.flysess", submission.checkpoint_id),
                 };
@@ -1487,8 +1554,5 @@ async fn write_one(shared: &Arc<WriterShared>, submission: &CaptureSubmission) -
             Err(e) => last = format!("the checkpoint writer stopped: {e}"),
         }
     }
-    SaveOutcome::Failed {
+    WriteOutcome::Failed { reason: last }
-        checkpoint_id: submission.checkpoint_id.clone(),
-        reason: last,
-    }
 }

services/flysim/crates/fly-session/tests/processes.rs

@@ -122,6 +122,7 @@ async fn a_slow_participant_is_resolved_rather_than_failed(mode: ExecutionMode)
         resolve_attempts: 4096,
         boot: Duration::from_secs(30),
         capture: Duration::from_secs(30),
+        durable: Duration::from_secs(60),
     };
     within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
     let reports = within("run", f.harness.coordinator.run(2))
@@ -193,6 +194,7 @@ async fn a_resolution_says_which_of_its_two_bounds_ended_it(mode: ExecutionMode)
         resolve_attempts: 8192,
         boot: Duration::from_secs(30),
         capture: Duration::from_secs(30),
+        durable: Duration::from_secs(60),
     };
     within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
     let started = Instant::now();
@@ -224,6 +226,7 @@ async fn a_resolution_says_which_of_its_two_bounds_ended_it(mode: ExecutionMode)
         resolve_attempts: 3,
         boot: Duration::from_secs(30),
         capture: Duration::from_secs(30),
+        durable: Duration::from_secs(60),
     };
     within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
     let failure = within("step", f.harness.coordinator.step())

services/flysim/crates/fly-session/tests/state.rs

@@ -15,6 +15,7 @@ mod common;
 use std::collections::BTreeSet;
 use std::path::Path;
 use std::sync::Arc;
+use std::time::Duration;
 
 use serde_json::{Value, json};
 
@@ -391,7 +392,8 @@ async fn a_lost_save_reply_holds_durable_metadata_in_every_mode(mode: ExecutionM
     a_lost_save_reply(Via::Unix, mode).await;
 }
 
-/// Two ways a save can end without a saved acknowledgment, and neither moves the mark.
+/// Three ways a save can end without a saved acknowledgment. None moves the mark, and each
+/// is reported as itself rather than as the others.
 async fn a_lost_save_reply(via: Via, mode: ExecutionMode) {
     let lost = ckpt(1);
     let config = HarnessConfig {
@@ -464,6 +466,59 @@ async fn a_lost_save_reply(via: Via, mode: ExecutionMode) {
     assert_eq!(resolved, None, "an unreferenced generation is never a restore candidate");
     assert_eq!(g.harness.coordinator.durable(), None);
     g.shutdown().await;
+
+    // The third: the caller's own budget runs out while the save is still going. That is a
+    // different fact from a lost reply -- this save has not stopped -- and it is reported as
+    // itself, because diagnosing one as the other is the implicit reading these rules refuse.
+    let slow = ckpt(3);
+    let gate = Arc::new(tokio::sync::Semaphore::new(0));
+    let mut config = HarnessConfig::default();
+    config.writer_faults.gate = Some(gate.clone());
+    let mut h = fx(via, mode, config).await;
+    within("bootstrap", h.harness.coordinator.bootstrap()).await.unwrap();
+    within("run", h.harness.coordinator.run(1)).await.unwrap();
+    // The durable budget is the caller's own and is not the call budget: this shortens the
+    // wait for an acknowledgment without shortening a single call to a participant.
+    h.harness.coordinator.deadlines.durable = Duration::from_millis(100);
+    let ticket = within("capture", h.harness.coordinator.capture(&slow, false))
+        .await
+        .unwrap();
+    let outcome = within("durable", h.harness.coordinator.await_durable(ticket))
+        .await
+        .unwrap();
+    assert_eq!(
+        outcome,
+        SaveOutcome::DeadlineExpired { checkpoint_id: slow.clone() },
+        "an expired caller budget is not a lost reply"
+    );
+    assert_ne!(outcome, SaveOutcome::ReplyLost { checkpoint_id: slow.clone() });
+    assert_eq!(h.harness.coordinator.durable(), None);
+    assert_eq!(
+        count(
+            &h.harness.coordinator.audit,
+            &format!("not-durable:failed:{slow}@1")
+        ),
+        1,
+        "the session records that this capture is not durable: {:?}",
+        h.harness.coordinator.audit
+    );
+
+    // It really was still going: once the writer is let past its gate the same operation
+    // commits, and resolving it is what moves the mark.
+    gate.add_permits(16);
+    let deadline = std::time::Instant::now() + Duration::from_secs(10);
+    let resolved = loop {
+        let found = within("resolve", h.harness.coordinator.resolve_durable(&slow))
+            .await
+            .unwrap();
+        if found.is_some() || std::time::Instant::now() >= deadline {
+            break found;
+        }
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    };
+    assert_eq!(resolved, Some(1), "the save the caller stopped waiting for still committed");
+    assert_eq!(h.harness.coordinator.durable(), Some((slow, 1)));
+    h.shutdown().await;
 }
 
 // ===============================================================================================