module Checkpoint.NativePeerPublication import Checkpoint.NativeRegions import Checkpoint.NativeEnvelope import Compiler.Planning.RegionFrame import Runtime.NativePhysicalProgram import Runtime.NativeLaunchRecipeLoad import Runtime.NativeMemoryPlan import Runtime.NativePeerEndpoint import Runtime.NativeRegionFrame import Representation.Partition import Runtime.ArenaCertificate import Std.Foundation import Model.Word64 import Std.List import Std.Natural family NativeRegionPaths : Type 0 constructor NativeRegionPathsOf field unrestricted nativeRegionFirstArgument : Nat field unrestricted nativeRegionPathCount : Nat end-family -- A star capture and durable publication strategy over owners 0..peers. -- State meaning, local computation, source placement and numerical policy -- belong to the caller. Owner 0 publishes: it prepares its own shards from -- its local state, then receives each other owner's captured state in turn -- over that owner's edge, checks that the agreed leading bytes (at least the -- (version, cursor, epoch) prefix) are identical to its own, and prepares -- that owner's shards from it. Only after -- every shard and the manifest are durable does it release each owner with -- the committed version; every owner, publisher included, completes only -- then. A missing, late or differently advanced owner publishes nothing. -- -- Per-owner frames carry that owner's identity, so a capture received on the -- wrong edge fails its frame check. All capture frames share one storage -- region (and all committed frames another): they are used one owner at a -- time, so the publisher's frame storage does not grow with the owner count. family NativePeerRegionPublication : Type 0 constructor NativePeerRegionPublicationOf field unrestricted peerPublicationCheckpoint : (family NativeRegionCheckpoint) -- owners other than the publisher field unrestricted peerPublicationPeers : Nat -- the frame carrying `owner`'s capture to the publisher, and the one releasing it field unrestricted peerPublicationCapture : (pi unrestricted owner : Nat . (family RegionFrameLayout)) field unrestricted peerPublicationCommitted : (pi unrestricted owner : Nat . (family RegionFrameLayout)) -- the connection slot to `peer` (Runtime.NativePeerEndpoint.nativePeerEdgeConnectionTo) field unrestricted peerPublicationConnection : (pi unrestricted peer : Nat . Nat) field unrestricted peerPublicationStateAt : Nat field unrestricted peerPublicationStateBytes : Nat -- the leading bytes of every owner's state that must equal the publisher's: -- at least the (version, cursor, epoch) capture prefix; a replicated learner -- names its whole replicated state, so diverged replicas publish nothing field unrestricted peerPublicationAgreedBytes : Nat -- which owner's state holds a shard, and the shard's byte offset in it field unrestricted peerPublicationOwner : (pi unrestricted shard : (family NativeRegionShard) . Nat) field unrestricted peerPublicationOffset : (pi unrestricted shard : (family NativeRegionShard) . Nat) field unrestricted peerPublicationPaths : (family NativeRegionPaths) end-family def nativeRegionPathFirst = (lambda unrestricted paths : (family NativeRegionPaths) . (eliminate NativeRegionPaths (lambda unrestricted current : (family NativeRegionPaths) . Nat) paths (branch NativeRegionPathsOf first count . first))) def nativeRegionPathCountOf = (lambda unrestricted paths : (family NativeRegionPaths) . (eliminate NativeRegionPaths (lambda unrestricted current : (family NativeRegionPaths) . Nat) paths (branch NativeRegionPathsOf first count . count))) def nativeRegionPathArgument = (lambda unrestricted index : Nat . (constructor NativePhysicalOperand NativePhysicalProcessArgument (modelWord64FromNaturalTruncated index))) def nativeRegionInputShardBase = (lambda unrestricted paths : (family NativeRegionPaths) . (succ (nativeRegionPathFirst paths))) def nativeRegionOutputShardBase = (lambda unrestricted paths : (family NativeRegionPaths) . (naturalAdd (nativeRegionInputShardBase paths) (nativeRegionPathCountOf paths))) def nativeRegionManifestCandidateIndex = (lambda unrestricted paths : (family NativeRegionPaths) . (naturalAdd (nativeRegionOutputShardBase paths) (naturalMultiply 2 (nativeRegionPathCountOf paths)))) def nativeRegionManifestCommittedIndex = (lambda unrestricted paths : (family NativeRegionPaths) . (succ (nativeRegionManifestCandidateIndex paths))) def nativeRegionDirectoryIndex = (lambda unrestricted paths : (family NativeRegionPaths) . (succ (nativeRegionManifestCommittedIndex paths))) def nativeRegionShardPath = (lambda unrestricted paths : (family NativeRegionPaths) . (lambda unrestricted index : Nat . (lambda unrestricted committed : Nat . (nativeRegionPathArgument (naturalAdd (nativeRegionOutputShardBase paths) (naturalAdd (naturalMultiply 2 index) committed)))))) -- Publication may include several durable file syncs. It has its own bounded -- 30-second readiness policy, independent of ordinary learning-message waits. def nativePeerPublicationIO = (nativePeerIOWithBudget (nativeLaunchRecipeImmediate 30000) (nativeLaunchRecipeImmediate nativePeerIOAttempts)) -- Prepare the shards held by `owner`, each from `stateBase` plus its offset, -- keeping each shard's index for its paths. Shards are independent: each has -- its own checkpoint layout, so grouping them by owner preserves the plan. def nativeRegionPrepareOwner = (lambda unrestricted plan : (family NativeRegionCheckpoint) . (lambda unrestricted captureAt : Nat . (lambda unrestricted ownerOf : (pi unrestricted shard : (family NativeRegionShard) . Nat) . (lambda unrestricted offsetOf : (pi unrestricted shard : (family NativeRegionShard) . Nat) . (lambda unrestricted owner : Nat . (lambda unrestricted stateBase : Nat . (lambda unrestricted paths : (family NativeRegionPaths) . (lambda unrestricted tail : (family NativePhysicalCommands) . (stdListFold (family NativeRegionShard) (pi unrestricted index : Nat . (family NativePhysicalCommands)) (lambda unrestricted shard : (family NativeRegionShard) . (lambda unrestricted next : (pi unrestricted index : Nat . (family NativePhysicalCommands)) . (lambda unrestricted index : Nat . (nat-eliminate (lambda unrestricted held : Nat . (family NativePhysicalCommands)) (next (succ index)) (lambda unrestricted previous : Nat . (lambda unrestricted ignored : (family NativePhysicalCommands) . (nativeRegionPrepareShard shard (nativeMemoryState captureAt) (nativeMemoryState (naturalAdd stateBase (offsetOf shard))) (nativeRegionShardPath paths index 0) (nativeRegionShardPath paths index 1) (nativeRegionPathArgument (nativeRegionDirectoryIndex paths)) nativePeerListenerSlot nativePeerScratchSlot (next (succ index))))) (naturalEqual (ownerOf shard) owner))))) (lambda unrestricted index : Nat . tail) (nativeRegionShards plan) 0))))))))) -- `body owner` for owners 1..peers, in order. def nativePeerEachOwner = (lambda unrestricted peers : Nat . (lambda unrestricted body : (pi unrestricted owner : Nat . (pi unrestricted rest : (family NativePhysicalCommands) . (family NativePhysicalCommands))) . (lambda unrestricted tail : (family NativePhysicalCommands) . (stdListFold Nat (family NativePhysicalCommands) (lambda unrestricted index : Nat . (lambda unrestricted rest : (family NativePhysicalCommands) . (body (succ index) rest))) tail (stdListNaturalsBelow peers))))) def nativePeerRegionPublish = (lambda unrestricted publication : (family NativePeerRegionPublication) . (lambda unrestricted role : Nat . (lambda unrestricted success : (family NativePhysicalCommands) . (eliminate NativePeerRegionPublication (lambda unrestricted current : (family NativePeerRegionPublication) . (family NativePhysicalCommands)) publication (branch NativePeerRegionPublicationOf plan peers capture committed connection stateAt stateBytes agreed ownerOf offsetOf paths . (let unrestricted version = (nativeMemoryLoad stateAt) in (let unrestricted epoch = (nativeMemoryLoad (naturalAdd stateAt (naturalMultiply 2 nativeCheckpointWordBytes))) in (nat-eliminate (lambda unrestricted current : Nat . (family NativePhysicalCommands)) (nativeRegionPrepareOwner plan stateAt ownerOf offsetOf 0 stateAt paths (nativePeerEachOwner peers (lambda unrestricted owner : Nat . (lambda unrestricted rest : (family NativePhysicalCommands) . (let unrestricted frame = (capture owner) in (nativeFrameInitializeRound frame epoch version (nativePeerPublicationIO (nativeLaunchRecipeSlotValue (connection owner)) 0 frame (nativeFrameConsume frame (nativeMemoryCompareRegions stateAt (framePayloadAt frame) agreed nativePeerScratchSlot b"checkpoint-captured-position" (nativeRegionPrepareOwner plan stateAt ownerOf offsetOf owner (framePayloadAt frame) paths rest)))))))) (nativeRegionPublish plan stateAt (lambda unrestricted index : Nat . (nativeRegionShardPath paths index 1)) (nativeRegionPathArgument (nativeRegionManifestCandidateIndex paths)) (nativeRegionPathArgument (nativeRegionManifestCommittedIndex paths)) (nativeRegionPathArgument (nativeRegionDirectoryIndex paths)) nativePeerListenerSlot nativePeerScratchSlot 6 (nativePeerEachOwner peers (lambda unrestricted owner : Nat . (lambda unrestricted rest : (family NativePhysicalCommands) . (let unrestricted released = (committed owner) in (nativeFrameInitializeRound released epoch version (nativeMemoryStoreWord64 b"checkpoint-committed-version" (framePayloadAt released) version (nativePeerPublicationIO (nativeLaunchRecipeSlotValue (connection owner)) 1 released rest)))))) success)))) (lambda unrestricted previous : Nat . (lambda unrestricted ignored : (family NativePhysicalCommands) . (let unrestricted channel = (nativeLaunchRecipeSlotValue (connection 0)) in (let unrestricted frame = (capture role) in (let unrestricted released = (committed role) in (nativeFrameInitializeRound frame epoch version (nativeFrameInitializeRound released epoch version (nativeMemoryCopyOperands (nativeMemoryState (framePayloadAt frame)) (nativeMemoryState stateAt) stateBytes nativePeerScratchSlot b"checkpoint-capture-state" (nativePeerPublicationIO channel 1 frame (nativePeerPublicationIO channel 0 released (nativeFrameConsume released (nativeLaunchRecipeAssertEqual (nativeMemoryLoad (framePayloadAt released)) version b"checkpoint-committed-version" success)))))))))))) role)))))))) -- Publication owns the frame and checkpoint scratch arenas. A shard's bytes -- must lie entirely within its owner's captured state payload, past the -- capture prefix, never in a header or digest workspace that publication -- itself overwrites. Offsets are relative to the owner's state. def nativePublicationSourceWithin = (lambda unrestricted start : Nat . (lambda unrestricted count : Nat . (lambda unrestricted base : Nat . (lambda unrestricted size : Nat . (naturalAnd (naturalLessOrEqual (naturalAdd base nativeRegionCaptureBytes) start) (naturalLessOrEqual (naturalAdd start count) (naturalAdd base size))))))) -- Every owner's frame occupies the first owner's storage exactly: same base, -- payload extent and end, so one arena resident covers all of them. def nativePeerFramesShareStorage = (lambda unrestricted peers : Nat . (lambda unrestricted frame : (pi unrestricted owner : Nat . (family RegionFrameLayout)) . (lambda unrestricted payload : Nat . (let unrestricted first = (frame 1) in (stdListFold Nat Nat (lambda unrestricted index : Nat . (lambda unrestricted rest : Nat . (let unrestricted current = (frame (succ index)) in (naturalAnd rest (naturalAnd (naturalEqual (framePayloadBytes current) payload) (naturalAnd (naturalEqual (frameBase current) (frameBase first)) (naturalEqual (frameEnd current) (frameEnd first)))))))) 1 (stdListNaturalsBelow peers)))))) def nativePeerPublicationAdmitted = (lambda unrestricted publication : (family NativePeerRegionPublication) . (lambda unrestricted available : Nat . (lambda unrestricted role : Nat . (eliminate NativePeerRegionPublication (lambda unrestricted current : (family NativePeerRegionPublication) . Nat) publication (branch NativePeerRegionPublicationOf plan peers capture committed connection stateAt stateBytes agreed ownerOf offsetOf paths . (naturalAnd (naturalNonzero peers) (naturalAnd (naturalLessOrEqual role peers) (naturalAnd (naturalBelowTwoPower 64 available) (naturalAnd (nativePeerFramesShareStorage peers capture stateBytes) (naturalAnd (nativePeerFramesShareStorage peers committed nativeCheckpointWordBytes) (naturalAnd (naturalAnd (naturalLess nativeRegionCaptureBytes stateBytes) (naturalAnd (naturalLessOrEqual nativeRegionCaptureBytes agreed) (naturalLessOrEqual agreed stateBytes))) (naturalAnd (naturalEqual (nativeRegionPathCountOf paths) (stdListLength (family NativeRegionShard) (nativeRegionShards plan))) (naturalAnd (arenaCertificate available (constructor ArenaResidents ArenaResidentsNext (constructor ArenaResident ArenaResidentValue b"captured-local-state" stateAt stateBytes nativeCheckpointWordBytes) (constructor ArenaResidents ArenaResidentsNext (constructor ArenaResident ArenaResidentValue b"capture-frame" (frameBase (capture 1)) (naturalSaturatingSubtract (frameEnd (capture 1)) (frameBase (capture 1))) 1) (constructor ArenaResidents ArenaResidentsNext (constructor ArenaResident ArenaResidentValue b"commit-frame" (frameBase (committed 1)) (naturalSaturatingSubtract (frameEnd (committed 1)) (frameBase (committed 1))) 1) (nativeRegionResidents plan))))) (stdListFold (family NativeRegionShard) Nat (lambda unrestricted shard : (family NativeRegionShard) . (lambda unrestricted rest : Nat . (naturalAnd rest (naturalAnd (naturalLessOrEqual (ownerOf shard) peers) (nativePublicationSourceWithin (offsetOf shard) (intervalExtent (nativeRegionInterval shard)) 0 stateBytes))))) 1 (nativeRegionShards plan))))))))))))))) def planNativePeerRegionPublish = (lambda unrestricted publication : (family NativePeerRegionPublication) . (lambda unrestricted available : Nat . (lambda unrestricted role : Nat . (lambda unrestricted success : (family NativePhysicalCommands) . (nat-eliminate (lambda unrestricted valid : Nat . (family StdResult (family NativeCheckpointFailure) (family NativePhysicalCommands))) (constructor StdResult StdFailure (family NativeCheckpointFailure) (family NativePhysicalCommands) (constructor NativeCheckpointFailure NativeCheckpointBounds)) (lambda unrestricted previous : Nat . (lambda unrestricted ignored : (family StdResult (family NativeCheckpointFailure) (family NativePhysicalCommands)) . (constructor StdResult StdSuccess (family NativeCheckpointFailure) (family NativePhysicalCommands) (nativePeerRegionPublish publication role success)))) (nativePeerPublicationAdmitted publication available role))))))