1 import Base 2 import ./change_tick.bend as Tick 3 import ./deferred_command.bend as Deferred 4 import ./event_stream.bend as Leaf 5 import ./world.bend as W 6 import ./relation_commands.bend as Notice 7 import ./relation_failure_runtime.bend as K 8 import ./internal/relation_failures/domains.bend as D 9 import ./internal/relation_failures/routing.bend as FailureRoute 10 import ./internal/relation_failures/barrier.bend as FailureBarrier 11 import ./event_runtime.bend as Ev 12 import ./event_family_runtime.bend as Family 13 import ./event_family_routing.bend as Route 14 15 # Descriptor metadata owns no second World. Typed publication ticks remain separate. 16 type Failures is Data: 17 Failures{domains:List<&2,D.Domain>>,positions:List<&2,Ev.Position>,tick:Nat,frameStart:Nat,boundary:Nat,capacity:Nat,nextReader:U32} 18 type State is Type: 19 State{typed:Family.State,failures:Failures} 20 21 def originals(~S:Data,~P:Data,records:List<&2,Notice.Notice>) -> List<&2,P>: 22 match records: 23 case []:[] 24 case Notice.Original{event} <> rest:event <> originals(~S,~P,rest) 25 case Notice.RelationFailed{_} <> rest:originals(~S,~P,rest) 26 case Notice.ComponentRemoved{_} <> rest:originals(~S,~P,rest) 27 case Notice.CleanupPaused{_} <> rest:originals(~S,~P,rest) 28 29 def started(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,typed:Family.State,runtime:K.Runtime>) -> Route.Runtime,State>: 30 match runtime: 31 case K.Runtime{world,domains,positions,tick,frameStart,boundary,capacity,nextReader}:Route.Runtime{world,State{typed,Failures{domains,positions,tick,frameStart,boundary,capacity,nextReader}},0n} 32 33 def start(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,world:W.World>,family:F,owners:H) -> Route.Runtime,State>: 34 started(~S,~C,~R,~P,~F,~H,Family.State{family,owners,[],0n},K.start(~S,~C,~R,~Notice.Notice,world)) 35 36 def synchronized(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,~sync:F -> Nat -> List<&2,P> -> F,state:State,+tick:Nat,observed:W.World> & List<&2,Notice.Notice>) -> W.World> & State: 37 match state: 38 case State{Family.State{family,owners,+journal,frameStart},failures}: 39 (world,records)=observed 40 +published=List.drop(&2,P,originals(~S,~P,records),Ev.size(~P,journal)) 41 (world,State{Family.State{sync(family,tick,published),owners,Ev.sync_checked(~P,Nat.is_lt(0n,Ev.count(~P,published)),journal,tick,published),frameStart},failures}) 42 43 def synchronize(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,~sync:F -> Nat -> List<&2,P> -> F,world:W.World>,state:State,tick:Nat) -> W.World> & State: 44 synchronized(~S,~C,~R,~P,~F,~H,~sync,state,tick,W.events_view(~S,~C,~R,~Notice.Notice,world)) 45 46 # The detached typed owner contains no World; K operations receive the sole World. 47 def failure_runtime(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,runtime:Route.Runtime,State>) -> K.Runtime> & (Family.State & Nat): 48 match runtime: 49 case Route.Runtime{world,State{typed,Failures{domains,positions,tick,frameStart,boundary,capacity,nextReader}},typedTick}:(K.Runtime{world,domains,positions,tick,frameStart,boundary,capacity,nextReader},(typed,typedTick)) 50 def failure_returned(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,held:Family.State & Nat,runtime:K.Runtime>) -> Route.Runtime,State>: 51 match held runtime: 52 case (typed,typedTick) K.Runtime{world,domains,positions,tick,frameStart,boundary,capacity,nextReader}: 53 Route.Runtime{world,State{typed,Failures{domains,positions,tick,frameStart,boundary,capacity,nextReader}},typedTick} 54 55 # Slots preserve the original mixed-log positions, including partially retained Publications. 56 type Retention is Type: 57 Retention{family:F,journal:List<&2,Ev.Batch

>,slots:List<&2,List<&2,P>>} 58 def retained_record(~P:Data,~F:Type,+head:List<&2,P>,returned:F & (List<&2,P> & List<&2,List<&2,P>>)) -> F & (List<&2,P> & List<&2,List<&2,P>>): 59 match returned: 60 case (family,(records,slots)):(family,(Ev.append(~P,head,records),head <> slots)) 61 def retained_record_head(~P:Data,~F:Type,rest:F -> F & (List<&2,P> & List<&2,List<&2,P>>),returned:F & List<&2,P>) -> F & (List<&2,P> & List<&2,List<&2,P>>): 62 match returned: 63 case (family,+head):retained_record(~P,~F,head,rest(family)) 64 def retained_records(~P:Data,~F:Type,~retain:F -> Nat -> List<&2,P> -> F & List<&2,P>,records:List<&2,P>,family:F,+tick:Nat) -> F & (List<&2,P> & List<&2,List<&2,P>>): 65 match records: 66 case []:(family,([],[])) 67 case head <> rest:retained_record_head(~P,~F,(family => retained_records(~P,~F,~retain,rest,family,tick)),retain(family,tick,[head])) 68 def retained_batch_join(~P:Data,~F:Type,tick:Nat,+records:List<&2,P>,slots:List<&2,List<&2,P>>,returned:Retention) -> Retention: 69 match returned: 70 case Retention{family,journal,tail}:Retention{family,Family.kept_batch(~P,Nat.is_lt(0n,Ev.count(~P,records)),tick,records,journal),List.append(&2,List<&2,P>,slots,tail)} 71 def retained_batch_head(~P:Data,~F:Type,rest:F -> Retention,tick:Nat,returned:F & (List<&2,P> & List<&2,List<&2,P>>)) -> Retention: 72 match returned: 73 case (family,(+records,slots)):retained_batch_join(~P,~F,tick,records,slots,rest(family)) 74 def retention(~P:Data,~F:Type,~retain:F -> Nat -> List<&2,P> -> F & List<&2,P>,journal:List<&2,Ev.Batch

>,family:F) -> Retention: 75 match journal: 76 case []:Retention{family,[],[]} 77 case Ev.Batch{+tick,records} <> rest:retained_batch_head(~P,~F,(family => retention(~P,~F,~retain,rest,family)),tick,retained_records(~P,~F,~retain,records,family,tick)) 78 def original_notices(~S:Data,~P:Data,records:List<&2,P>) -> List<&2,Notice.Notice>: 79 match records: 80 case []:[] 81 case head <> rest:Notice.Original{head} <> original_notices(~S,~P,rest) 82 def rebuilt(~S:Data,~P:Data,notices:List<&2,Notice.Notice>,slots:List<&2,List<&2,P>>) -> List<&2,Notice.Notice>: 83 match notices slots: 84 case Nil{} _:[] 85 case Con{Notice.Original{_},rest} Nil{}:rebuilt(~S,~P,rest,[]) 86 case Con{Notice.Original{_},rest} Con{head,tail}:List.append(&2,Notice.Notice,original_notices(~S,~P,head),rebuilt(~S,~P,rest,tail)) 87 case Con{Notice.RelationFailed{failure},rest} slots:Notice.RelationFailed{failure} <> rebuilt(~S,~P,rest,slots) 88 case Con{Notice.ComponentRemoved{id},rest} slots:Notice.ComponentRemoved{id} <> rebuilt(~S,~P,rest,slots) 89 case Con{Notice.CleanupPaused{frames},rest} slots:Notice.CleanupPaused{frames} <> rebuilt(~S,~P,rest,slots) 90 def framed(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,owners:H,failures:Failures,+tick:Nat,observed:W.World> & List<&2,Notice.Notice>,returned:Retention) -> Route.Runtime,State>: 91 match observed returned: 92 case (world,notices) Retention{family,journal,slots}: 93 Route.Runtime{W.events_replace(~S,~C,~R,~Notice.Notice,world,rebuilt(~S,~P,notices,slots)),State{Family.State{family,owners,journal,tick},failures},tick} 94 def frame(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,~frame:F -> Nat -> F,~retain:F -> Nat -> List<&2,P> -> F & List<&2,P>,runtime:Route.Runtime,State>) -> Route.Runtime,State>: 95 match runtime: 96 case Route.Runtime{world,State{Family.State{family,owners,journal,_},failures},+tick}:framed(~S,~C,~R,~P,~F,~H,owners,failures,tick,W.events_view(~S,~C,~R,~Notice.Notice,world),retention(~P,~F,~retain,journal,frame(family,tick))) 97 98 def failure_framed(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,detached:K.Runtime> & (Family.State & Nat)) -> Route.Runtime,State>: 99 match detached: 100 case (runtime,held):failure_returned(~S,~C,~R,~P,~F,~H,held,K.frame(~S,~C,~R,~Notice.Notice,runtime)) 101 def frame_failures(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,runtime:Route.Runtime,State>) -> Route.Runtime,State>: 102 failure_framed(~S,~C,~R,~P,~F,~H,failure_runtime(~S,~C,~R,~P,~F,~H,runtime)) 103 def inventoried(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,world:W.World>,owners:H,journal:List<&2,Ev.Batch

>,frameStart:Nat,failures:Failures,tick:Nat,returned:F & List<&2,String>) -> Route.Runtime,State> & Maybe<&2,List<&2,String>>: 104 match returned: 105 case (family,names):(Route.Runtime{world,State{Family.State{family,owners,journal,frameStart},failures},tick},Some{names}) 106 def inventory_enabled(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,~inventory:F -> F & List<&2,String>,enabled:Bool,runtime:Route.Runtime,State>) -> Route.Runtime,State> & Maybe<&2,List<&2,String>>: 107 match enabled runtime: 108 case False{} runtime:(runtime,None{}) 109 case True{} Route.Runtime{world,State{Family.State{family,owners,journal,frameStart},failures},tick}:inventoried(~S,~C,~R,~P,~F,~H,world,owners,journal,frameStart,failures,tick,inventory(family)) 110 111 def selected(~S:Data,~E:Data,~P:Data,~F:Type,~H:Type,failures:Failures,returned:Family.State & Leaf.Stream) -> State & Leaf.Stream: 112 match returned: 113 case (typed,stream):(State{typed,failures},stream) 114 def take(~S:Data,~E:Data,~P:Data,~F:Type,~H:Type,~take:F -> F & Leaf.Stream,state:State) -> State & Leaf.Stream: 115 match state: 116 case State{typed,failures}:selected(~S,~E,~P,~F,~H,failures,Family.take(~S,~E,~P,~F,~H,~take,typed)) 117 def put(~S:Data,~E:Data,~P:Data,~F:Type,~H:Type,~put:F -> Leaf.Stream -> F,state:State,stream:Leaf.Stream) -> State: 118 match state: 119 case State{typed,failures}:State{Family.put(~S,~E,~P,~F,~H,~put,typed,stream),failures} 120 121 # Reserve from the same failure-reader pool used by standalone K.register. 122 def reserved_failure_reader(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,held:Family.State & Nat,result:K.Runtime> & Maybe<&2,U32>) -> Route.Runtime,State> & Maybe<&2,U32>: 123 match result: 124 case (runtime,id):(failure_returned(~S,~C,~R,~P,~F,~H,held,runtime),id) 125 def reserve_failure_reader_detached(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,detached:K.Runtime> & (Family.State & Nat)) -> Route.Runtime,State> & Maybe<&2,U32>: 126 match detached: 127 case (runtime,held):reserved_failure_reader(~S,~C,~R,~P,~F,~H,held,K.reserve_reader(~S,~C,~R,~Notice.Notice,runtime)) 128 def reserve_failure_reader(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,runtime:Route.Runtime,State>) -> Route.Runtime,State> & Maybe<&2,U32>: 129 reserve_failure_reader_detached(~S,~C,~R,~P,~F,~H,failure_runtime(~S,~C,~R,~P,~F,~H,runtime)) 130 131 132 # One outer ingestion boundary, shared by every parameter in an invocation. 133 def ingested_failures(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,typed:Family.State,positions:List<&2,Ev.Position>,+tick:Nat,start:Nat,boundary:Nat,capacity:Nat,nextReader:U32,returned:W.World> & List<&2,D.Domain>>) -> Route.Runtime,State>: 134 match returned: 135 case (world,domains):Route.Runtime{world,State{typed,Failures{domains,positions,tick,start,boundary,capacity,nextReader}},tick} 136 def ingested_typed(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,~classify:Notice.Notice -> Maybe>,+tick:Nat,returned:W.World> & State) -> Route.Runtime,State>: 137 match returned: 138 case (world,State{typed,Failures{domains,positions,_,start,boundary,capacity,nextReader}}): 139 ingested_failures(~S,~C,~R,~P,~F,~H,typed,positions,tick,start,boundary,capacity,nextReader,FailureRoute.collect(~S,~C,~R,~Notice.Notice,~classify,world,tick,domains)) 140 def ingest(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,~sync:F -> Nat -> List<&2,P> -> F,~classify:Notice.Notice -> Maybe>,+tick:Nat,runtime:Route.Runtime,State>) -> Route.Runtime,State>: 141 match runtime: 142 case Route.Runtime{world,state,_}:ingested_typed(~S,~C,~R,~P,~F,~H,~classify,tick,synchronize(~S,~C,~R,~P,~F,~H,~sync,world,state,tick)) 143 def attempt_clocked(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,state:State,tick:Nat,clocked:W.World> & Tick.Tick) -> Route.Runtime,State>: 144 match clocked: 145 case (world,_):Route.Runtime{world,state,1n+tick} 146 def attempt_ingested(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,runtime:Route.Runtime,State>) -> Route.Runtime,State>: 147 match runtime: 148 case Route.Runtime{world,state,tick}:attempt_clocked(~S,~C,~R,~P,~F,~H,state,tick,W.execution_clock(~S,~C,~R,~Notice.Notice,world)) 149 # Standalone and composed operations share histories but may advance different 150 # administrative clocks. Never publish below any already observed boundary. 151 def before_state(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,~sync:F -> Nat -> List<&2,P> -> F,~classify:Notice.Notice -> Maybe>,state:State,+tick:Nat,world:W.World>,clock:Tick.Tick) -> Route.Runtime,State>: 152 match state: 153 case State{typed,Failures{domains,positions,+failureTick,start,boundary,capacity,nextReader}}: 154 attempt_ingested(~S,~C,~R,~P,~F,~H,ingest(~S,~C,~R,~P,~F,~H,~sync,~classify,Nat.max(1n,Nat.max(tick,Nat.max(failureTick,Tick.total(clock)))),Route.Runtime{world,State{typed,Failures{domains,positions,failureTick,start,boundary,capacity,nextReader}},tick})) 155 def before_clocked(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,~sync:F -> Nat -> List<&2,P> -> F,~classify:Notice.Notice -> Maybe>,state:State,tick:Nat,clocked:W.World> & Tick.Tick) -> Route.Runtime,State>: 156 match clocked: 157 case (world,clock):before_state(~S,~C,~R,~P,~F,~H,~sync,~classify,state,tick,world,clock) 158 # Zero is the initial unread cursor, so startup publications need a positive epoch. 159 # Called only after registration/parameter validation; preparation follows it. 160 def before_attempt(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,~sync:F -> Nat -> List<&2,P> -> F,~classify:Notice.Notice -> Maybe>,runtime:Route.Runtime,State>) -> Route.Runtime,State>: 161 match runtime: 162 case Route.Runtime{world,state,tick}:before_clocked(~S,~C,~R,~P,~F,~H,~sync,~classify,state,tick,W.clock(~S,~C,~R,~Notice.Notice,world)) 163 # Called once after successful cursor commits, without another clock advance. 164 def after_success(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,~sync:F -> Nat -> List<&2,P> -> F,~classify:Notice.Notice -> Maybe>,runtime:Route.Runtime,State>) -> Route.Runtime,State>: 165 match runtime: 166 case Route.Runtime{world,state,+tick}:ingest(~S,~C,~R,~P,~F,~H,~sync,~classify,tick,Route.Runtime{world,state,tick}) 167 168 169 # Internal invocation transport: detached metadata retains all shared owners. 170 def split(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,runtime:Route.Runtime,State>) -> W.World> & (State & Nat): 171 match runtime: 172 case Route.Runtime{world,state,tick}:(world,(state,tick)) 173 def join(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,world:W.World>,metadata:State & Nat) -> Route.Runtime,State>: 174 match metadata: 175 case (state,tick):Route.Runtime{world,state,tick} 176 177 # The command-aware barrier keeps each command publication as its own batch. 178 # Typed owners/journal, reader positions and membership remain unchanged. 179 def barrier_drained(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,typed:Family.State,positions:List<&2,Ev.Position>,frameStart:Nat,boundary:Nat,capacity:Nat,nextReader:U32,returned:FailureBarrier.BarrierState>) -> Route.Runtime,State>: 180 match returned: 181 case FailureBarrier.BarrierState{world,domains,+tick}:Route.Runtime{world,State{typed,Failures{domains,positions,tick,frameStart,boundary,capacity,nextReader}},tick} 182 183 def barrier_advanced(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,~classify:Notice.Notice -> Maybe>,typed:Family.State,domains:List<&2,D.Domain>>,positions:List<&2,Ev.Position>,previous:Nat,frameStart:Nat,boundary:Nat,capacity:Nat,nextReader:U32,advanced:W.World> & Tick.Tick) -> Route.Runtime,State>: 184 match advanced: 185 case (W.World{namespace,allocator,store,resource,events,pending,registrations,nextSystemId,clock},actual): 186 barrier_drained(~S,~C,~R,~P,~F,~H,typed,positions,frameStart,boundary,capacity,nextReader, 187 FailureBarrier.drain(~S,~C,~R,~Notice.Notice,~classify,List.reverse(&1,Deferred.Command>>,pending), 188 FailureBarrier.BarrierState{W.World{namespace,allocator,store,resource,events,[],registrations,nextSystemId,clock},domains,Nat.max(previous,Tick.total(actual))})) 189 190 def barrier(~S:Data,~C:Type,~R:Type,~P:Data,~F:Type,~H:Type,~classify:Notice.Notice -> Maybe>,context:Route.Runtime,State>) -> Route.Runtime,State>: 191 match context: 192 case Route.Runtime{world,State{typed,Failures{domains,positions,failureTick,frameStart,boundary,capacity,nextReader}},tick}: 193 barrier_advanced(~S,~C,~R,~P,~F,~H,~classify,typed,domains,positions,Nat.max(tick,failureTick),frameStart,boundary,capacity,nextReader,W.enter_run(~S,~C,~R,~Notice.Notice,world))