diff --git a/docs/design/session-framework/checkpoint-envelope-v1.md b/docs/design/session-framework/checkpoint-envelope-v1.md index 7cb833f..d58ed98 100644 --- a/docs/design/session-framework/checkpoint-envelope-v1.md +++ b/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. diff --git a/services/flysim/crates/fly-session-types/examples/update_fixtures.rs b/services/flysim/crates/fly-session-types/examples/update_fixtures.rs index d68b865..bad3693 100644 --- a/services/flysim/crates/fly-session-types/examples/update_fixtures.rs +++ b/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": [], diff --git a/services/flysim/crates/fly-session-types/fixtures/checkpoint-envelope.json b/services/flysim/crates/fly-session-types/fixtures/checkpoint-envelope.json index 70453c3..b726585 100644 --- a/services/flysim/crates/fly-session-types/fixtures/checkpoint-envelope.json +++ b/services/flysim/crates/fly-session-types/fixtures/checkpoint-envelope.json @@ -59,8 +59,8 @@ ], "admissionState": null, "eventWatermarks": { - "lastEventId": "evt-1", - "lastOrdinal": "7" + "lastSourceStep": "42", + "issued": "7" } }, "environment": { @@ -119,49 +119,49 @@ } ], "envelope": { - "base64": "RkxZU0VTUzEBAAAAIAAAADUIAAAFAAAAWAgAAAAAAAB7ImFnZW50cyI6W3siYWdlbnRJZCI6ImZseS1hIiwiYnJhaW5UaWNrcyI6IjI1MzQiLCJkYXRhc2V0RGlnZXN0IjoiNmMwYWYxZjA3ODRlZjYzYTM5M2VlNzdkNjE0ZTgyNDZjNjI1MDUxMzYwZjNmMWE0ODgzODM3NGM1ZDM1NWI1MiIsIm1vZGVsVmVyc2lvbiI6ImxpZi0xbXMtZjY0LXYyIiwicGF5bG9hZCI6ImFnZW50LWZseS1hIiwicGxhc3RpY2l0eVZlcnNpb24iOiJmbHkta2MtbWJvbi1yc3RkcC12MiIsInByb2ZpbGVEaWdlc3QiOiIxOTAwZWFiNmMwMjg0ODNkNzEyNjU5OWVlNmY1MGRlMGQyNzkwN2I1YzY1ZmE5MDUyNDU4MGI0YjBmOTg1MmIwIiwicmVtYWluZGVyIjp7ImRlbm9taW5hdG9yIjoiMyIsIm51bWVyYXRvciI6IjEwMDAwMDAifSwic2VlZCI6LTE4NDk0NjA2M31dLCJjaGVja3BvaW50SWQiOiJja3B0LTEiLCJjb21wYXRpYmlsaXR5Ijp7ImJhY2tlbmREaWdlc3QiOiIxMGUwOGE0MTllODUwZWJhMWViYmExOGZkZDI4ZWI3ZWMxYjdlOGJhYTliY2MzYjk3M2UyYjg4OTFlYzcyNmJlIiwiY29udGVudERpZ2VzdCI6ImVkNzAwMmI0MzllOWFjODQ1ZjIyMzU3ZDgyMmJhYzE0NDQ3MzBmYmRiNjAxNmQzZWM5NDMyMjk3YjllYzlmNzMiLCJjb250cm9sbGVyRGlnZXN0IjoiYzE0NzIxMzViMTRjNzdjOGJlZjk4ZTczZjcwMjA4MzI1ZmEwZGNmMWU2YmQ2NjhhZTliMzFhOWNlYTI5NWZlNyIsInBhcnNlckRpZ2VzdCI6ImIxN2Q0NTEyMTE1MDkyOGYyMTQ2YWY0OWUxOTVlZmYxZWVmNWQ2NzMyNWJlMjczYTczM2ZiNzRhY2FkYWEzNDIiLCJwYXRjaERpZ2VzdCI6ImE0ODk1ZWI0NGFmYzMzNmZlY2JiYTZlNTIwY2Q2N2UxNzhkYWNlMDI3NjY1NWQxMDJmY2VmZmE4ZTVmNzA1NzAiLCJzdGF0ZUZvcm1hdElkIjoiZmx5c2Vzcy0xIn0sImNvbXBvc2l0aW9uRGlnZXN0IjoiNzMwZDcyNWM4YTU5ZDNhNzMwM2RlZjJiZWQwNDFhNTc3ZWRiNDI1NWFhYmQ0ODg5Y2UxMjkxODMxMWQ5NTJmMCIsImNvb3JkaW5hdG9yIjp7ImFkbWlzc2lvblN0YXRlIjpudWxsLCJldmVudFdhdGVybWFya3MiOnsibGFzdEV2ZW50SWQiOiJldnQtMSIsImxhc3RPcmRpbmFsIjoiNyJ9LCJleGVjdXRvclN0YXRlIjpbeyJhZ2VudElkIjoiZmx5LWEiLCJwYXlsb2FkIjoiZXhlY3V0b3ItZmx5LWEifV0sInByaW9ySW5zcGVjdGlvbiI6InByaW9yLWluc3BlY3Rpb24iLCJ0YXNrTGVkZ2VyIjoidGFzay1sZWRnZXIifSwiZW52ZWxvcGVWZXJzaW9uIjoxLCJlbnZpcm9ubWVudCI6eyJwYXlsb2FkIjoid29ybGQiLCJ3b3JrZXJJZCI6ImFyZW5hIn0sImVwaXNvZGVJZCI6ImVwaXNvZGUtMSIsImhlbHBlclN0YXRlIjpbXSwicGF5bG9hZHMiOlt7ImJ5dGVMZW5ndGgiOiIxNyIsImRpZ2VzdCI6IjEzMjFkZmZiMGNkYzZmOTA5MmNiZjdmYTJhNWZjNjhiYmVkMTJjOTkzZDVhZDM5ODI2NDAxMjgxMGNlOWJmOTMiLCJuYW1lIjoiYWdlbnQtZmx5LWEifSx7ImJ5dGVMZW5ndGgiOiIxNCIsImRpZ2VzdCI6IjNhZWU2MGRmN2UyOWVmZWJhN2Y1Zjk5ZmM1ODY3NjQ3YjM2YWViZmYxZDVkM2M4MzhkYmZmMzIzMTIyZTY0NjIiLCJuYW1lIjoiZXhlY3V0b3ItZmx5LWEifSx7ImJ5dGVMZW5ndGgiOiIxMSIsImRpZ2VzdCI6IjQwYjAwZWQyYmJiYTkwMWQ2ODIwNWZmNzFiMDRhNDRiOWVlNTNjNTFjYjMxMDlhYTJjZWFhNDRmMWM0NTcyN2UiLCJuYW1lIjoidGFzay1sZWRnZXIifSx7ImJ5dGVMZW5ndGgiOiIxMCIsImRpZ2VzdCI6IjJjMTNiN2I0ZDlhOTkxNjgwMWFiOTE5MWMzMTRmMzFiMDQ1ZTliOWM1YjY2OWE2YzA0NzRmMDIxN2VmNzViZjUiLCJuYW1lIjoicHJpb3ItaW5zcGVjdGlvbiJ9LHsiYnl0ZUxlbmd0aCI6IjY0IiwiZGlnZXN0IjoiZjVhNWZkNDJkMTZhMjAzMDI3OThlZjZlZDMwOTk3OWI0MzAwM2QyMzIwZDlmMGU4ZWE5ODMxYTkyNzU5ZmI0YiIsIm5hbWUiOiJ3b3JsZCJ9XSwicG9ydE1hcCI6W3siYWdlbnRJZCI6ImZseS1hIiwicG9ydElkIjoicG9ydC0xIn1dLCJzY2hlZHVsZXJJZCI6ImxvY2tzdGVwLXYxIiwic291cmNlU2NvcGUiOnsiZXBvY2giOiJlcG9jaC0xIiwic2Vzc2lvbklkIjoiZGVtbyIsInN0ZXAiOiI0MiJ9LCJ3b3JsZFRpbWUiOnsiZGVub21pbmF0b3IiOiIxIiwibnVtZXJhdG9yIjoiNzAwMDAwMDAwIn19AAAAYWdlbnQtZmx5LWEAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAIgKAAAAAAAAEQAAAAAAAAATId/7DNxvkJLL9/oqX8aLvtEsmT1a05gmQBKBDOm/k2V4ZWN1dG9yLWZseS1hAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAACgCgAAAAAAAA4AAAAAAAAAOu5g334p7+un9fmfxYZ2R7Nq6/8dXTyDjb/zIxIuZGJ0YXNrLWxlZGdlcgAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAsAoAAAAAAAALAAAAAAAAAECwDtK7upAdaCBf9xsEpEue5TxRyzEJqizqpE8cRXJ+cHJpb3ItaW5zcGVjdGlvbgAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAMAKAAAAAAAACgAAAAAAAAAsE7e02amRaAGrkZHDFPMbBF6bnFtmmmwEdPAhfvdb9XdvcmxkAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAADQCgAAAAAAAEAAAAAAAAAA9aX9QtFqIDAnmO9u0wmXm0MAPSMg2fDo6pgxqSdZ+0thZ2VudCBzdGF0ZSBieXRlcwAAAAAAAABleGVjdXRvciBzdGF0ZQAAeyJyYW5rIjoxMH0AAAAAAHsibWFwIjo0MH0AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAQAsAAAAAAAAhBlh9AWLTtVKckmeIzNn4DO5Yhn2C1nUA1T1RfxMXlkZMWVNFU1NG", - "byteLength": 2880, + "base64": "RkxZU0VTUzEBAAAAIAAAADAIAAAFAAAAUAgAAAAAAAB7ImFnZW50cyI6W3siYWdlbnRJZCI6ImZseS1hIiwiYnJhaW5UaWNrcyI6IjI1MzQiLCJkYXRhc2V0RGlnZXN0IjoiNmMwYWYxZjA3ODRlZjYzYTM5M2VlNzdkNjE0ZTgyNDZjNjI1MDUxMzYwZjNmMWE0ODgzODM3NGM1ZDM1NWI1MiIsIm1vZGVsVmVyc2lvbiI6ImxpZi0xbXMtZjY0LXYyIiwicGF5bG9hZCI6ImFnZW50LWZseS1hIiwicGxhc3RpY2l0eVZlcnNpb24iOiJmbHkta2MtbWJvbi1yc3RkcC12MiIsInByb2ZpbGVEaWdlc3QiOiIxOTAwZWFiNmMwMjg0ODNkNzEyNjU5OWVlNmY1MGRlMGQyNzkwN2I1YzY1ZmE5MDUyNDU4MGI0YjBmOTg1MmIwIiwicmVtYWluZGVyIjp7ImRlbm9taW5hdG9yIjoiMyIsIm51bWVyYXRvciI6IjEwMDAwMDAifSwic2VlZCI6LTE4NDk0NjA2M31dLCJjaGVja3BvaW50SWQiOiJja3B0LTEiLCJjb21wYXRpYmlsaXR5Ijp7ImJhY2tlbmREaWdlc3QiOiIxMGUwOGE0MTllODUwZWJhMWViYmExOGZkZDI4ZWI3ZWMxYjdlOGJhYTliY2MzYjk3M2UyYjg4OTFlYzcyNmJlIiwiY29udGVudERpZ2VzdCI6ImVkNzAwMmI0MzllOWFjODQ1ZjIyMzU3ZDgyMmJhYzE0NDQ3MzBmYmRiNjAxNmQzZWM5NDMyMjk3YjllYzlmNzMiLCJjb250cm9sbGVyRGlnZXN0IjoiYzE0NzIxMzViMTRjNzdjOGJlZjk4ZTczZjcwMjA4MzI1ZmEwZGNmMWU2YmQ2NjhhZTliMzFhOWNlYTI5NWZlNyIsInBhcnNlckRpZ2VzdCI6ImIxN2Q0NTEyMTE1MDkyOGYyMTQ2YWY0OWUxOTVlZmYxZWVmNWQ2NzMyNWJlMjczYTczM2ZiNzRhY2FkYWEzNDIiLCJwYXRjaERpZ2VzdCI6ImE0ODk1ZWI0NGFmYzMzNmZlY2JiYTZlNTIwY2Q2N2UxNzhkYWNlMDI3NjY1NWQxMDJmY2VmZmE4ZTVmNzA1NzAiLCJzdGF0ZUZvcm1hdElkIjoiZmx5c2Vzcy0xIn0sImNvbXBvc2l0aW9uRGlnZXN0IjoiNzMwZDcyNWM4YTU5ZDNhNzMwM2RlZjJiZWQwNDFhNTc3ZWRiNDI1NWFhYmQ0ODg5Y2UxMjkxODMxMWQ5NTJmMCIsImNvb3JkaW5hdG9yIjp7ImFkbWlzc2lvblN0YXRlIjpudWxsLCJldmVudFdhdGVybWFya3MiOnsiaXNzdWVkIjoiNyIsImxhc3RTb3VyY2VTdGVwIjoiNDIifSwiZXhlY3V0b3JTdGF0ZSI6W3siYWdlbnRJZCI6ImZseS1hIiwicGF5bG9hZCI6ImV4ZWN1dG9yLWZseS1hIn1dLCJwcmlvckluc3BlY3Rpb24iOiJwcmlvci1pbnNwZWN0aW9uIiwidGFza0xlZGdlciI6InRhc2stbGVkZ2VyIn0sImVudmVsb3BlVmVyc2lvbiI6MSwiZW52aXJvbm1lbnQiOnsicGF5bG9hZCI6IndvcmxkIiwid29ya2VySWQiOiJhcmVuYSJ9LCJlcGlzb2RlSWQiOiJlcGlzb2RlLTEiLCJoZWxwZXJTdGF0ZSI6W10sInBheWxvYWRzIjpbeyJieXRlTGVuZ3RoIjoiMTciLCJkaWdlc3QiOiIxMzIxZGZmYjBjZGM2ZjkwOTJjYmY3ZmEyYTVmYzY4YmJlZDEyYzk5M2Q1YWQzOTgyNjQwMTI4MTBjZTliZjkzIiwibmFtZSI6ImFnZW50LWZseS1hIn0seyJieXRlTGVuZ3RoIjoiMTQiLCJkaWdlc3QiOiIzYWVlNjBkZjdlMjllZmViYTdmNWY5OWZjNTg2NzY0N2IzNmFlYmZmMWQ1ZDNjODM4ZGJmZjMyMzEyMmU2NDYyIiwibmFtZSI6ImV4ZWN1dG9yLWZseS1hIn0seyJieXRlTGVuZ3RoIjoiMTEiLCJkaWdlc3QiOiI0MGIwMGVkMmJiYmE5MDFkNjgyMDVmZjcxYjA0YTQ0YjllZTUzYzUxY2IzMTA5YWEyY2VhYTQ0ZjFjNDU3MjdlIiwibmFtZSI6InRhc2stbGVkZ2VyIn0seyJieXRlTGVuZ3RoIjoiMTAiLCJkaWdlc3QiOiIyYzEzYjdiNGQ5YTk5MTY4MDFhYjkxOTFjMzE0ZjMxYjA0NWU5YjljNWI2NjlhNmMwNDc0ZjAyMTdlZjc1YmY1IiwibmFtZSI6InByaW9yLWluc3BlY3Rpb24ifSx7ImJ5dGVMZW5ndGgiOiI2NCIsImRpZ2VzdCI6ImY1YTVmZDQyZDE2YTIwMzAyNzk4ZWY2ZWQzMDk5NzliNDMwMDNkMjMyMGQ5ZjBlOGVhOTgzMWE5Mjc1OWZiNGIiLCJuYW1lIjoid29ybGQifV0sInBvcnRNYXAiOlt7ImFnZW50SWQiOiJmbHktYSIsInBvcnRJZCI6InBvcnQtMSJ9XSwic2NoZWR1bGVySWQiOiJsb2Nrc3RlcC12MSIsInNvdXJjZVNjb3BlIjp7ImVwb2NoIjoiZXBvY2gtMSIsInNlc3Npb25JZCI6ImRlbW8iLCJzdGVwIjoiNDIifSwid29ybGRUaW1lIjp7ImRlbm9taW5hdG9yIjoiMSIsIm51bWVyYXRvciI6IjcwMDAwMDAwMCJ9fWFnZW50LWZseS1hAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAACACgAAAAAAABEAAAAAAAAAEyHf+wzcb5CSy/f6Kl/Gi77RLJk9WtOYJkASgQzpv5NleGVjdXRvci1mbHktYQAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAmAoAAAAAAAAOAAAAAAAAADruYN9+Ke/rp/X5n8WGdkezauv/HV08g42/8yMSLmRidGFzay1sZWRnZXIAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAKgKAAAAAAAACwAAAAAAAABAsA7Su7qQHWggX/cbBKRLnuU8UcsxCaos6qRPHEVyfnByaW9yLWluc3BlY3Rpb24AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAC4CgAAAAAAAAoAAAAAAAAALBO3tNmpkWgBq5GRwxTzGwRem5xbZppsBHTwIX73W/V3b3JsZAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAyAoAAAAAAABAAAAAAAAAAPWl/ULRaiAwJ5jvbtMJl5tDAD0jINnw6OqYMaknWftLYWdlbnQgc3RhdGUgYnl0ZXMAAAAAAAAAZXhlY3V0b3Igc3RhdGUAAHsicmFuayI6MTB9AAAAAAB7Im1hcCI6NDB9AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAADgLAAAAAAAAX4L9WkdX0MViD3h5YJgf7VQgwocWU4XJoKX6MUwl+8hGTFlTRVNTRg==", + "byteLength": 2872, "layout": { "headerBytes": 32, "manifestOffset": "32", - "manifestBytes": 2101, - "tableOffset": "2136", + "manifestBytes": 2096, + "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" } ] diff --git a/services/flysim/crates/fly-session/README.md b/services/flysim/crates/fly-session/README.md index 33f0594..76b8b72 100644 --- a/services/flysim/crates/fly-session/README.md +++ b/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 - save reply is `SaveOutcome::ReplyLost`, and `Coordinator::resolve_durable` then asks the - store about the *same* checkpoint instead of saving again. -- **The writer is bounded twice**, by outstanding captures and by queued bytes, and it 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. + happened, which is the durable commit point. Only the second moves the durable mark, and the + three ways it can end without one are told apart: `Failed` (the write stopped), + `ReplyLost` (the write finished and the acknowledgment did not arrive) and + `DeadlineExpired` (the caller's own budget ran out while the save was still going). + `Coordinator::resolve_durable` then asks the store about the *same* checkpoint instead of + 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 diff --git a/services/flysim/crates/fly-session/src/coordinator.rs b/services/flysim/crates/fly-session/src/coordinator.rs index bde1183..f01d72b 100644 --- a/services/flysim/crates/fly-session/src/coordinator.rs +++ b/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 { 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) } diff --git a/services/flysim/crates/fly-session/src/state.rs b/services/flysim/crates/fly-session/src/state.rs index 93ba77b..aa42d3c 100644 --- a/services/flysim/crates/fly-session/src/state.rs +++ b/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 - /// durable metadata does not move; the caller resolves the *same* operation against the - /// store manifest instead of saving again. + /// The writer finished this job and its reply channel was gone before the outcome could + /// be delivered. The write is over and its result is unknown from here. 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> { 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 { - queue - .iter() - .position(|job| job.submission.replaceable) - .map(|index| queue.remove(index).expect("just found")) + let replaced_index = if submission.replaceable { + queue.iter().position(|job| job.submission.replaceable) } 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 - /// an outcome and not a hang. + /// Waits for one save's outcome, within the caller's own budget. + /// + /// 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, 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 { + 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) { 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 { - SaveOutcome::Committed { .. } => stats.committed += 1, - SaveOutcome::Failed { .. } | SaveOutcome::ReplyLost { .. } => stats.failed += 1, - SaveOutcome::Superseded { .. } => stats.superseded += 1, + match &written { + WriteOutcome::Committed { .. } => stats.committed += 1, + WriteOutcome::Failed { .. } => stats.failed += 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) { } } -async fn write_one(shared: &Arc, submission: &CaptureSubmission) -> SaveOutcome { +async fn write_one(shared: &Arc, 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 { - checkpoint_id: submission.checkpoint_id.clone(), + return WriteOutcome::Failed { reason: format!("payload {}: {}", payload.name, e.message), }; } }; if bytes.len() as u64 != payload.byte_length || digest_of_bytes(&bytes) != payload.digest { - return SaveOutcome::Failed { - checkpoint_id: submission.checkpoint_id.clone(), + return WriteOutcome::Failed { reason: format!( "payload {} is not the content its capture declared", payload.name @@ -1442,8 +1511,7 @@ async fn write_one(shared: &Arc, submission: &CaptureSubmission) - let bytes = match checkpoint::encode(&submission.manifest, &payloads) { Ok(bytes) => bytes, Err(e) => { - return SaveOutcome::Failed { - checkpoint_id: submission.checkpoint_id.clone(), + return WriteOutcome::Failed { reason: format!("envelope: {}", e.0), }; } @@ -1477,8 +1545,7 @@ async fn write_one(shared: &Arc, submission: &CaptureSubmission) - .await; match result { Ok(Ok(())) => { - return SaveOutcome::Committed { - checkpoint_id: submission.checkpoint_id.clone(), + return WriteOutcome::Committed { boundary: submission.boundary, file: format!("{}.flysess", submission.checkpoint_id), }; @@ -1487,8 +1554,5 @@ async fn write_one(shared: &Arc, submission: &CaptureSubmission) - Err(e) => last = format!("the checkpoint writer stopped: {e}"), } } - SaveOutcome::Failed { - checkpoint_id: submission.checkpoint_id.clone(), - reason: last, - } + WriteOutcome::Failed { reason: last } } diff --git a/services/flysim/crates/fly-session/tests/processes.rs b/services/flysim/crates/fly-session/tests/processes.rs index 17dd0e2..0eaa267 100644 --- a/services/flysim/crates/fly-session/tests/processes.rs +++ b/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()) diff --git a/services/flysim/crates/fly-session/tests/state.rs b/services/flysim/crates/fly-session/tests/state.rs index d7b3a0f..b287c04 100644 --- a/services/flysim/crates/fly-session/tests/state.rs +++ b/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; } // ===============================================================================================