owned_event_family_reader_runtime.bend
bendvy-ecs-core@0.1.0.1 · 72 lines · 0 definitions
1import Base
2import bendvy-ecs-core@0.1.0.1/trace_facts.bend as Facts
3import bendvy-ecs-core@0.1.0.1/world.bend as W
4import bendvy-ecs-core@0.1.0.1/system.bend as Sys
5import bendvy-ecs-core@0.1.0.1/event_runtime.bend as Ev
6import bendvy-ecs-core@0.1.0.1/transaction.bend as T
7import bendvy-ecs-core@0.1.0.1/capabilities.bend as Cap
8import bendvy-ecs-core@0.1.0.1/owned_event_log.bend as Log
9import bendvy-ecs-core@0.1.0.1/owned_event_family_storage.bend as Storage
10import bendvy-ecs-core@0.1.0.1/owner_product.bend as Product
11import bendvy-ecs-core@0.1.0.1/event_stream.bend as Stream
12import bendvy-ecs-core@0.1.0.1/event_family_routing.bend as Route
13import bendvy-ecs-core@0.1.0.1/ordinary_owned_event.bend as Ord
14
15# One declared reader folds its unread payloads through a confined request cap.
16# Providers are trusted schema provisioning; payload ownership stays in the log.
17type Failure<Error:Data> is Data:
18 BodyFailed{error:Error}
19 MissingPayload{}
20type Reply<Error:Data,Args:Type> is Type:
21 Reply{outcome:T.Outcome<Failure<Error>>,args:Args,lagged:Bool}
22def stopped(~S:Data,~C:Type,~R:Type,~P:Type,~Rest:Type,~E:Data,~path:Product.Lens<R,Log.Log<P>,Rest>,~Args:Type,~Error:Data,world:W.World<S,C,R,E>,args:Args,outcome:T.Outcome<Failure<Error>>,lagged:Bool) -> W.World<S,C,R,E> & Reply<Error,Args>:
23 (world,Reply{outcome,args,lagged})
24def visited(~S:Data,~C:Type,~R:Type,~P:Type,~Rest:Type,~E:Data,~path:Product.Lens<R,Log.Log<P>,Rest>,~Args:Type,~Error:Data,lagged:Bool,next:W.World<S,C,R,E> -> Args -> W.World<S,C,R,E> & Reply<Error,Args>,result:W.World<S,C,R,E> & Log.ReadResult<Args,T.Outcome<Error> & Args>) -> W.World<S,C,R,E> & Reply<Error,Args>:
25 match result:
26 case (world,Log.Missing{args}):stopped(~S,~C,~R,~P,~Rest,~E,~path,~Args,~Error,world,args,T.Failure{MissingPayload{}},lagged)
27 case (world,Log.Found{(T.Failure{error},args)}):stopped(~S,~C,~R,~P,~Rest,~E,~path,~Args,~Error,world,args,T.Failure{BodyFailed{error}},lagged)
28 case (world,Log.Found{(T.Success{},args)}):next(world,args)
29def fold_rows(~S:Data,~C:Type,~R:Type,~P:Type,~Rest:Type,~E:Data,~path:Product.Lens<R,Log.Log<P>,Rest>,~Request:Type,~Projection:Data,~Args:Type,~Error:Data,~at:P -> Request -> P & Projection,~body:@-H:Type -> @-read:Cap.Request<H,Request,Projection> -> H -> Args -> H & (T.Outcome<Error> & Args),keys:List<&2,Nat>,world:W.World<S,C,R,E>,args:Args,+lagged:Bool) -> W.World<S,C,R,E> & Reply<Error,Args>:
30 match keys:
31 case []:stopped(~S,~C,~R,~P,~Rest,~E,~path,~Args,~Error,world,args,T.Success{},lagged)
32 case key <> remaining:visited(~S,~C,~R,~P,~Rest,~E,~path,~Args,~Error,lagged,nextWorld => nextArgs => fold_rows(~S,~C,~R,~P,~Rest,~E,~path,~Request,~Projection,~Args,~Error,~at,~body,remaining,nextWorld,nextArgs,lagged),Storage.read(~S,~C,~R,~P,~Rest,~E,~path,~Request,~Projection,~Args,~(T.Outcome<Error> & Args),~at,~body,world,key,args))
33def transported(~S:Data,~C:Type,~R:Type,~P:Type,~Rest:Type,~E:Data,~path:Product.Lens<R,Log.Log<P>,Rest>,~Args:Type,~Error:Data,result:W.World<S,C,R,E> & Reply<Error,Args>) -> Sys.Outcome<W.World<S,C,R,E>,Unit,Reply<Error,Args>>:
34 match result:
35 case (world,reply):Sys.Succeeded{world,reply}
36def runner(~S:Data,~C:Type,~R:Type,~P:Type,~Rest:Type,~E:Data,~path:Product.Lens<R,Log.Log<P>,Rest>,~Request:Type,~Projection:Data,~Args:Type,~Error:Data,~at:P -> Request -> P & Projection,~body:@-H:Type -> @-read:Cap.Request<H,Request,Projection> -> H -> Args -> H & (T.Outcome<Error> & Args),world:W.World<S,C,R,E>,input:Ev.Input<Nat,Args>) -> Sys.Outcome<W.World<S,C,R,E>,Unit,Reply<Error,Args>>:
37 match input:
38 case Ev.Input{keys,lagged,args}:transported(~S,~C,~R,~P,~Rest,~E,~path,~Args,~Error,fold_rows(~S,~C,~R,~P,~Rest,~E,~path,~Request,~Projection,~Args,~Error,~at,~body,keys,world,args,lagged))
39
40# The closed runner retains affine Args even on semantic failure. Route.run must
41# see transport success; reconcile its selected stream to the exact existing
42# failed-reader activation rule afterward, without changing other streams.
43def restored_stream(~S:Data,old:List<&2,Ev.Position>,id:U32,tick:Nat,stream:Stream.Stream<S,Nat>) -> Stream.Stream<S,Nat>:
44 match stream:
45 case Stream.Stream{domain,batches,_,nextReader,frameStart,boundary,capacity,dropped}:Stream.Stream{domain,batches,Ev.activate(old,id,Nat.sub(tick,1n)),nextReader,frameStart,boundary,capacity,dropped}
46def restore_taken(~S:Data,~C:Type,~R:Type,~E:Data,~F:Type,~put:F -> Stream.Stream<S,Nat> -> F,world:W.World<S,C,R,E>,+tick:Nat,old:List<&2,Ev.Position>,id:U32,returned:F & Stream.Stream<S,Nat>) -> Route.Runtime<S,C,R,E,F>:
47 match returned:
48 case (family,stream):Route.Runtime{world,put(family,restored_stream(~S,old,id,tick,stream)),tick}
49def restore(~S:Data,~C:Type,~R:Type,~E:Data,~F:Type,~take:F -> F & Stream.Stream<S,Nat>,~put:F -> Stream.Stream<S,Nat> -> F,old:List<&2,Ev.Position>,id:U32,runtime:Route.Runtime<S,C,R,E,F>) -> Route.Runtime<S,C,R,E,F>:
50 match runtime:
51 case Route.Runtime{world,family,tick}:restore_taken(~S,~C,~R,~E,~F,~put,world,tick,old,id,take(family))
52def reconciled(~S:Data,~C:Type,~R:Type,~P:Type,~Rest:Type,~E:Data,~path:Product.Lens<R,Log.Log<P>,Rest>,~Request:Type,~Projection:Data,~Args:Type,~Error:Data,~at:P -> Request -> P & Projection,~body:@-H:Type -> @-read:Cap.Request<H,Request,Projection> -> H -> Args -> H & (T.Outcome<Error> & Args),~F:Type,~take:F -> F & Stream.Stream<S,Nat>,~put:F -> Stream.Stream<S,Nat> -> F,old:List<&2,Ev.Position>,id:U32,result:Route.RunResult<S,C,R,E,Nat,F,Args,Unit,Reply<Error,Args>,take,put,runner(~S,~C,~R,~P,~Rest,~E,~path,~Request,~Projection,~Args,~Error,~at,~body)>) -> Route.RunResult<S,C,R,E,Nat,F,Args,Unit,Reply<Error,Args>,take,put,runner(~S,~C,~R,~P,~Rest,~E,~path,~Request,~Projection,~Args,~Error,~at,~body)>:
53 match result:
54 case Route.Ran{runtime,reader,registry,Ev.Succeeded{Reply{T.Failure{error},args,lagged}}}:Route.Ran{restore(~S,~C,~R,~E,~F,~take,~put,old,id,runtime),reader,registry,Ev.Succeeded{Reply{T.Failure{error},args,lagged}}}
55 case result:result
56def run_taken(~S:Data,~C:Type,~R:Type,~P:Type,~Rest:Type,~E:Data,~path:Product.Lens<R,Log.Log<P>,Rest>,~Request:Type,~Projection:Data,~Args:Type,~Error:Data,~at:P -> Request -> P & Projection,~body:@-H:Type -> @-read:Cap.Request<H,Request,Projection> -> H -> Args -> H & (T.Outcome<Error> & Args),~F:Type,~take:F -> F & Stream.Stream<S,Nat>,~put:F -> Stream.Stream<S,Nat> -> F,~synchronize:W.World<S,C,R,E> -> F -> Nat -> W.World<S,C,R,E> & F,world:W.World<S,C,R,E>,tick:Nat,reader:Route.Reader<S,Nat,F,take,put>,registry:Sys.Registry<S,C,R,E,Ev.Input<Nat,Args>,Unit,Reply<Error,Args>,runner(~S,~C,~R,~P,~Rest,~E,~path,~Request,~Projection,~Args,~Error,~at,~body)>,args:Args,returned:F & Stream.Stream<S,Nat>) -> Route.RunResult<S,C,R,E,Nat,F,Args,Unit,Reply<Error,Args>,take,put,runner(~S,~C,~R,~P,~Rest,~E,~path,~Request,~Projection,~Args,~Error,~at,~body)>:
57 match reader returned:
58 case Route.Reader{namespace,+id,registryId}, (family,Stream.Stream{_,_,positions,_,_,_,_,_}):reconciled(~S,~C,~R,~P,~Rest,~E,~path,~Request,~Projection,~Args,~Error,~at,~body,~F,~take,~put,positions,id,Route.run(~S,~C,~R,~E,~Nat,~F,~Args,~Unit,~Reply<Error,Args>,~take,~put,~runner(~S,~C,~R,~P,~Rest,~E,~path,~Request,~Projection,~Args,~Error,~at,~body),~synchronize,Route.Runtime{world,family,tick},Route.Reader{namespace,id,registryId},registry,args))
59def run(~S:Data,~C:Type,~R:Type,~P:Type,~Rest:Type,~E:Data,~path:Product.Lens<R,Log.Log<P>,Rest>,~Request:Type,~Projection:Data,~Args:Type,~Error:Data,~at:P -> Request -> P & Projection,~body:@-H:Type -> @-read:Cap.Request<H,Request,Projection> -> H -> Args -> H & (T.Outcome<Error> & Args),~F:Type,~take:F -> F & Stream.Stream<S,Nat>,~put:F -> Stream.Stream<S,Nat> -> F,~synchronize:W.World<S,C,R,E> -> F -> Nat -> W.World<S,C,R,E> & F,runtime:Route.Runtime<S,C,R,E,F>,reader:Route.Reader<S,Nat,F,take,put>,registry:Sys.Registry<S,C,R,E,Ev.Input<Nat,Args>,Unit,Reply<Error,Args>,runner(~S,~C,~R,~P,~Rest,~E,~path,~Request,~Projection,~Args,~Error,~at,~body)>,args:Args) -> Route.RunResult<S,C,R,E,Nat,F,Args,Unit,Reply<Error,Args>,take,put,runner(~S,~C,~R,~P,~Rest,~E,~path,~Request,~Projection,~Args,~Error,~at,~body)>:
60 match runtime:
61 case Route.Runtime{world,family,tick}:run_taken(~S,~C,~R,~P,~Rest,~E,~path,~Request,~Projection,~Args,~Error,~at,~body,~F,~take,~put,~synchronize,world,tick,reader,registry,args,take(family))
62def runner_observed(~S:Data,~C:Type,~R:Type,~P:Type,~Rest:Type,~E:Data,~path:Product.Lens<R,Log.Log<P>,Rest>,~Request:Type,~Projection:Data,~Args:Type,~Error:Data,~at:P -> Request -> P & Projection,~body:@-H:Type -> @-read:Cap.Request<H,Request,Projection> -> H -> Args -> H & (T.Outcome<Error> & Args),trace:Facts.State,world:W.World<S,C,R,E>,input:Ev.Input<Nat,Args>) -> Sys.Outcome<W.World<S,C,R,E>,Unit,Reply<Error,Args>> & Facts.State:
63 (runner(~S,~C,~R,~P,~Rest,~E,~path,~Request,~Projection,~Args,~Error,~at,~body,world,input),trace)
64def accepted(~Error:Data,~Args:Type,value:Reply<Error,Args>) -> Reply<Error,Args> & Bool:
65 match value:
66 case Reply{T.Success{},args,lagged}:(Reply{T.Success{},args,lagged},True{})
67 case Reply{T.Failure{error},args,lagged}:(Reply{T.Failure{error},args,lagged},False{})
68def run_traced(~S:Data,~C:Type,~R:Type,~P:Type,~Rest:Type,~E:Data,~path:Product.Lens<R,Log.Log<P>,Rest>,~Request:Type,~Projection:Data,~Args:Type,~Error:Data,~at:P -> Request -> P & Projection,~body:@-H:Type -> @-read:Cap.Request<H,Request,Projection> -> H -> Args -> H & (T.Outcome<Error> & Args),~F:Type,~take:F -> F & Stream.Stream<S,Nat>,~put:F -> Stream.Stream<S,Nat> -> F,~synchronize:W.World<S,C,R,E> -> F -> Nat -> W.World<S,C,R,E> & F,runtime:Route.Runtime<S,C,R,E,F>,reader:Route.Reader<S,Nat,F,take,put>,registry:Sys.Registry<S,C,R,E,Ev.Input<Nat,Args>,Unit,Reply<Error,Args>,runner(~S,~C,~R,~P,~Rest,~E,~path,~Request,~Projection,~Args,~Error,~at,~body)>,args:Args,trace:Facts.State) -> Route.RunResult<S,C,R,E,Nat,F,Args,Unit,Reply<Error,Args>,take,put,runner(~S,~C,~R,~P,~Rest,~E,~path,~Request,~Projection,~Args,~Error,~at,~body)> & Facts.State:
69 Route.run_traced(~S,~C,~R,~E,~Nat,~F,~Args,~Unit,~Reply<Error,Args>,~take,~put,~runner(~S,~C,~R,~P,~Rest,~E,~path,~Request,~Projection,~Args,~Error,~at,~body),~runner_observed(~S,~C,~R,~P,~Rest,~E,~path,~Request,~Projection,~Args,~Error,~at,~body),~accepted(~Error,~Args),~synchronize,runtime,reader,registry,args,trace)
70
71def register(~S:Data,~C:Type,~R:Type,~P:Type,~Rest:Type,~E:Data,~path:Product.Lens<R,Log.Log<P>,Rest>,~Request:Type,~Projection:Data,~Args:Type,~Error:Data,~at:P -> Request -> P & Projection,~body:@-H:Type -> @-read:Cap.Request<H,Request,Projection> -> H -> Args -> H & (T.Outcome<Error> & Args),~F:Type,~take:F -> F & Stream.Stream<S,Nat>,~put:F -> Stream.Stream<S,Nat> -> F,runtime:Route.Runtime<S,C,R,E,F>,name:String) -> Route.Registration<S,C,R,E,Nat,F,Args,Unit,Reply<Error,Args>,take,put,runner(~S,~C,~R,~P,~Rest,~E,~path,~Request,~Projection,~Args,~Error,~at,~body)>:
72 Route.register(~S,~C,~R,~E,~Nat,~F,~Args,~Unit,~Reply<Error,Args>,~take,~put,~runner(~S,~C,~R,~P,~Rest,~E,~path,~Request,~Projection,~Args,~Error,~at,~body),runtime,name)