fix(stream): bound subscriber delivery
Cap agents queues before mailboxes and coalesce session wakes. Slow readers overflow cleanly while model work keeps moving.
143 unmodified linesExpand all14414514614714814915015115222 unmodified lines175176177178179180181182183211 unmodified linesExpand all143 unmodified linesExpand all self.assertEqual( [m["name"] for m in self.api(f"/sessions/{parent}/children")], ["coder"] )
@exclusive def test_tree_snapshot_and_event_stream(self): parent = self.app.session() heard = self.listen() child = self.spawn(parent)["session"]["id"]22 unmodified lines wait_for(lambda: expected <= kinds(), timeout=10) except AssertionError: self.fail(f"missing stream events: {expected - kinds()}; got {kinds()}")
@exclusive def test_rename_keeps_family_address_and_reports_stream_event(self): parent = self.app.session() child = self.spawn(parent)["session"]["id"] self.app.idle(child)211 unmodified linesExpand all123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145%% Inspect real stream owners while publishers fill an unread TCP connection.-module(albedo_stream_pressure_probe).-export([pressure_json/1, subscriptions_json/0, removed_json/0, session_pressure_json/1, session_removed_json/1, burst_json/3]).
rows(Table) -> case ets:whereis(Table) of undefined -> []; _ -> ets:tab2list(Table) end.
bytes(Binary) when is_binary(Binary) -> byte_size(Binary);bytes(Tuple) when is_tuple(Tuple) -> bytes(tuple_to_list(Tuple));bytes([Head | Tail]) -> bytes(Head) + bytes(Tail);bytes(_) -> 0.
blocked(Owner) -> case process_info(Owner, current_stacktrace) of {current_stacktrace, Frames} -> case lists:any(fun({_, Function, _, _}) -> lists:member(Function, [send, send_event, port_command]) end, Frames) of true -> 1; false -> 0 end; undefined -> 0 end.
sample({Ref, {Owner, _, Control}}) -> {Count, Bytes} = case ets:lookup(albedo_bus_queues, Ref) of [{_, {_, C, B}}] -> {C, B}; [] -> {0, 0} end, case process_info(Owner, [messages, message_queue_len]) of undefined -> #{count => Count, bytes => Bytes, mailbox => 0, mailbox_bytes => 0, blocked => 0, overflow => atomics:get(Control, 1)}; Info -> #{count => Count, bytes => Bytes, mailbox => proplists:get_value(message_queue_len, Info), mailbox_bytes => bytes(proplists:get_value(messages, Info)), blocked => blocked(Owner), overflow => atomics:get(Control, 1)} end.
measure() -> lists:foldl(fun(Row, Peak) -> maps:merge_with(fun(_, A, B) -> max(A, B) end, Peak, sample(Row)) end, #{count => 0, bytes => 0, mailbox => 0, mailbox_bytes => 0, blocked => 0, overflow => 0}, rows(albedo_bus)).
publish(Ids, Start, Count, Size) -> lists:foreach(fun(Index) -> Id = lists:nth(1 + ((Index - 1) rem length(Ids)), Ids), Text = iolist_to_binary([<<"pressure:">>, Id, <<":">>, integer_to_binary(Index), <<":">>, binary:copy(<<"x">>, Size)]), Event = iolist_to_binary([<<"{\"type\":\"text\",\"text\":\"">>, Text, <<"\"}">>]), 'albedo@daemon@bus':activity(Id, Event) end, lists:seq(Start, Start + Count - 1)), nil.
publish_batch(Ids, Published) -> Workers = [spawn_monitor(fun() -> publish([Id], Published div length(Ids) + 1, 12, 4096) end) || Id <- Ids], lists:foreach(fun({Worker, Monitor}) -> receive {'DOWN', Monitor, process, Worker, normal} -> ok; {'DOWN', Monitor, process, Worker, Reason} -> error({publisher_failed, Reason}) after 10000 -> error(publisher_blocked) end end, Workers).
pressure_json(Ids) -> io_lib:format("~s", [json:encode(pressure(Ids, 0, measure()))]).
pressure(Ids, Published, Peak) when Published < 10000 -> publish_batch(Ids, Published), Current = measure(), Next = maps:merge_with(fun(_, A, B) -> max(A, B) end, Peak, Current), case maps:get(overflow, Current) > 0 andalso maps:get(blocked, Next) > 0 of true -> Next#{published => Published + 96, pressure => true}; false -> %% Pace below a responsive subscriber's per-flush queue budget. receive after 150 -> ok end, pressure(Ids, Published + 96, Next) end;pressure(_, Published, Peak) -> Peak#{published => Published, pressure => false}.
burst_json(Ids, Count, Size) -> [{_, {Owner, _, _}}] = rows(albedo_bus), %% Hold the real stream handler at a barrier while testing each budget. ok = sys:suspend(Owner), try Peak = lists:foldl(fun(_, Acc) -> publish(Ids, 1, 1, Size), maps:merge_with(fun(_, A, B) -> max(A, B) end, Acc, measure()) end, measure(), lists:seq(1, Count)), json:encode(Peak) after sys:resume(Owner) end.
subscriptions_json() -> json:encode(#{subscribers => length(rows(albedo_bus)), queues => length(rows(albedo_bus_queues))}).
removed_json() -> removed(1000).removed(0) -> subscriptions_json();removed(Left) -> case {rows(albedo_bus), rows(albedo_bus_queues)} of {[], []} -> subscriptions_json(); _ -> receive after 10 -> ok end, removed(Left - 1) end.
session_state(Id) -> {some, Session} = 'albedo@daemon@session':live(Id), {ok, Owner} = 'gleam@erlang@process':subject_owner(Session), sys:get_state(Owner).
wakes(wake) -> 1;wakes(Tuple) when is_tuple(Tuple) -> wakes(tuple_to_list(Tuple));wakes([Head | Tail]) -> wakes(Head) + wakes(Tail);wakes(_) -> 0.
session_sample(Id) -> State = session_state(Id), Watchers = element(14, State), Samples = [case process_info(element(2, Watcher), messages) of undefined -> #{wakes => 0, blocked => 0}; {messages, Messages} -> #{wakes => wakes(Messages), blocked => blocked(element(2, Watcher))} end || Watcher <- Watchers], Peak = lists:foldl(fun(Sample, Acc) -> maps:merge_with(fun(_, A, B) -> max(A, B) end, Acc, Sample) end, #{wakes => 0, blocked => 0}, Samples), Peak#{watchers => length(Watchers), sequence => element(12, State)}.
session_pressure_json(Id) -> json:encode(session_pressure(Id, 1500, #{})).session_pressure(Id, Left, Peak) -> Sample = session_sample(Id), Next = maps:merge_with(fun(_, A, B) -> max(A, B) end, Peak, Sample), Complete = maps:get(blocked, Next) > 0 andalso not 'albedo@daemon@bus':is_running(Id), case Complete orelse Left =:= 0 of true -> Next; false -> receive after 10 -> ok end, session_pressure(Id, Left - 1, Next) end.
session_removed_json(Id) -> json:encode(session_removed(Id, 1000)).session_removed(Id, Left) -> Sample = session_sample(Id), case maps:get(watchers, Sample) =:= 0 orelse Left =:= 0 of true -> Sample; false -> receive after 10 -> ok end, session_removed(Id, Left - 1) end.123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231"""Unread TCP streams cannot retain unbounded activity or block model completion."""
import jsonimport socketimport subprocessimport threadingimport unittest
from harness import Albedo, Provider, ROOT, exclusive, text
class StreamProbe: def __init__(self, app): self.app = app source = ROOT / "test/e2e/albedo_stream_pressure_probe.erl" subprocess.run( ["erlc", "-Werror", "-o", str(app.root), str(source)], check=True, timeout=30, )
def call(self, function, arguments="[]"): app = self.app cookie = (app.home / "inspect.cookie").read_text().strip() node = f"albedo_{app.daemon._pid}@{socket.gethostname().split('.')[0]}" beam = app.root / "albedo_stream_pressure_probe.beam" expression = ( f"Node = '{node}', " f"{{ok, Binary}} = file:read_file({json.dumps(str(beam))}), " "{module, albedo_stream_pressure_probe} = " "rpc:call(Node, code, load_binary, " '[albedo_stream_pressure_probe, "probe.erl", Binary]), ' f"io:put_chars(rpc:call(Node, albedo_stream_pressure_probe, {function}, {arguments})), halt()." ) result = subprocess.run( [ "erl", "+S", "2:2", "-sname", f"pressure_probe_{app.daemon._pid}", "-setcookie", cookie, "-noshell", "-eval", expression, ], capture_output=True, text=True, check=False, timeout=40, ) if result.returncode != 0: raise AssertionError(result.stdout + result.stderr) return json.loads(result.stdout)
def unread_stream(app, path): connection = socket.socket() connection.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024) connection.settimeout(15) connection.connect(("127.0.0.1", app.connection["port"])) connection.sendall( ( f"GET {path} HTTP/1.1\r\nHost: localhost\r\n" f"Authorization: Bearer {app.connection['token']}\r\n\r\n" ).encode() ) # Read only the readiness batch, then leave all later socket data unread. initial = b"" while b"data: " not in initial or b"\n\n" not in initial.split(b"data: ", 1)[1]: part = connection.recv(1) if not part: raise AssertionError("stream closed before readiness") initial += part return connection
@exclusiveclass StreamPressureTest(unittest.TestCase): def setUp(self): self.provider = Provider(lambda _: text("model completed under pressure")) self.addCleanup(self.provider.close) self.app = Albedo( self.provider, prepare=lambda app: app.daemon.env.update(ALBEDO_INSPECT="1"), ) self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None) self.probe = StreamProbe(self.app)
def test_stalled_agents_feed_is_bounded_while_responsive_feed_and_model_complete( self, ): sessions = [self.app.session() for _ in range(8)] stalled = unread_stream(self.app, "/agents/stream") self.addCleanup(stalled.close) observed = [] ready = threading.Event() completed = threading.Event()
def consume(): with self.app.api("/agents/stream") as response: for line in response: if line.startswith(b"data: "): events = json.loads(line[6:])["events"] observed.extend(events) ready.set() if any( event.get("text") == "model completed under pressure" for event in events ): completed.set() return
threading.Thread(target=consume, daemon=True).start() self.assertTrue(ready.wait(5), "responsive feed never subscribed") ids = "[" + ",".join(f"<<{json.dumps(session)}>>" for session in sessions) + "]" measured = self.probe.call("pressure_json", f"[{ids}]") self.assertTrue(measured["pressure"], measured) self.assertEqual(measured["blocked"], 1, measured) self.app.prompt( sessions[0], "complete while another subscriber is stalled" ).close() self.app.idle(sessions[0]) self.assertTrue(completed.wait(5), "responsive feed lost model completion") self.assertLessEqual(measured["count"], 256, measured) self.assertLessEqual(measured["bytes"], 1048576, measured) self.assertLessEqual(measured["mailbox_bytes"], 1024, measured) self.assertLessEqual(measured["mailbox"], 3, measured) pressure_events = [ event for event in observed if event.get("text", "").startswith("pressure:") ] self.assertEqual(len(pressure_events), measured["published"]) for session in sessions: self.assertEqual( [ int(event["text"].split(":", 3)[2]) for event in pressure_events if event["session"] == session ], list(range(1, measured["published"] // len(sessions) + 1)), ) self.assertFalse(any(event["type"] == "overflow" for event in observed)) with self.app.api(f"/agents?session={sessions[0]}") as response: snapshot = json.load(response) self.assertFalse(snapshot["nodes"][0]["running"]) stalled.close() self.assertEqual( self.probe.call("removed_json"), {"subscribers": 0, "queues": 0} )
def test_idle_agents_disconnect_removes_subscription_without_publication(self): connection = unread_stream(self.app, "/agents/stream") self.assertEqual(self.probe.call("subscriptions_json")["subscribers"], 1) connection.close() self.assertEqual( self.probe.call("removed_json"), {"subscribers": 0, "queues": 0} )
def test_agent_count_byte_and_oversize_limits_emit_only_overflow_then_close(self): session = self.app.session() for name, count, size in ( ("count", 257, 16), ("bytes", 128, 16384), ("oversize", 1, 1048577), ): with self.subTest(limit=name), self.app.api("/agents/stream") as response: initial = next(line for line in response if line.startswith(b"data: ")) self.assertEqual(json.loads(initial[6:])["events"], []) measured = self.probe.call( "burst_json", f"[[<<{json.dumps(session)}>>],{count},{size}]" ) batches = [ json.loads(line[6:]) for line in response if line.startswith(b"data: ") ] self.assertEqual(batches, [{"events": [{"type": "overflow"}]}]) self.assertEqual(measured["overflow"], 1, measured) self.assertLessEqual(measured["count"], 256, measured) self.assertLessEqual(measured["bytes"], 1048576, measured) if name == "count": self.assertEqual(measured["count"], 256, measured) elif name == "bytes": self.assertLess(measured["count"], 256, measured) else: self.assertEqual(measured["bytes"], 0, measured) self.assertEqual( self.probe.call("removed_json"), {"subscribers": 0, "queues": 0} ) self.app.prompt(session, "model continues after overflow").close() self.app.idle(session) with self.app.api(f"/agents?session={session}") as response: snapshot = json.load(response) self.assertFalse(snapshot["nodes"][0]["running"])
def test_stalled_session_has_one_wake_and_reconnect_resets_durable_history(self): session = self.app.session() before = self.app.stream_page(session) connection = unread_stream(self.app, f"/sessions/{session}/stream") self.addCleanup(connection.close) self.provider.chunk_size = 4096 self.provider.script = lambda _: text("session pressure " * 250000, delay=0.002) self.app.prompt(session, "durable prompt under session pressure").close() measured = self.probe.call( "session_pressure_json", f"[<<{json.dumps(session)}>>]" ) self.assertEqual(measured["blocked"], 1, measured) self.assertLessEqual(measured["wakes"], 1, measured) self.app.idle(session) recovered = self.app.stream_page(session, before, tail=2) self.assertEqual(recovered["generation"], before["generation"]) self.assertEqual(recovered["events"][0]["type"], "reset") self.assertIn( "durable prompt under session pressure", [ event.get("text") for event in recovered["events"] if event["type"] == "user" ], )
def test_idle_session_disconnect_removes_watcher_without_publication(self): session = self.app.session() connection = unread_stream(self.app, f"/sessions/{session}/stream") connection.close() measured = self.probe.call( "session_removed_json", f"[<<{json.dumps(session)}>>]" ) self.assertEqual(measured["watchers"], 0, measured)12341236456789101112101113141516171819202122232425262728294 unmodified lines34353637242526272838394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341358 unmodified lines%% The agents bus: every session's activity, fanned out to whoever watches the%% orchestrator view. Subscribers are closures owned by a process; a closure%% only sends a message, and a dead owner's entry is dropped on the next%% publish, so a closed stream needs no cleanup call.%% Bounded subscriber queues. Publishers admit directly into ETS; subscriber%% mailboxes carry only a coalesced wake, never event payloads.-module(albedo_bus).-export([subscribe/2, publish/1, has_subscribers/0, mark_running/2, running/1, forget/1]).-export([subscribe/2, drain/1, rearm/2, publish/1, has_subscribers/0, mark_running/2, running/1, forget/1]).
-define(TABLE, albedo_bus).-define(QUEUES, albedo_bus_queues).-define(EVENT_LIMIT, 256).-define(BYTE_LIMIT, 1048576).-define(CAS_ATTEMPTS, 32).
subscribe(Owner, Fun) -> albedo_registry:register(?TABLE, make_ref(), {Owner, Fun}).subscribe(Owner, Notify) -> Ref = make_ref(), Control = atomics:new(2, [{signed, false}]), albedo_registry:register(?QUEUES, Ref, {queue:new(), 0, 0}), albedo_registry:register(?TABLE, Ref, {Owner, Notify, Control}), spawn(fun() -> Monitor = erlang:monitor(process, Owner), receive {'DOWN', Monitor, process, Owner, _} -> ets:delete(?TABLE, Ref), ets:delete(?QUEUES, Ref) end end), Ref.
has_subscribers() -> case ets:info(?TABLE, size) of Size when is_integer(Size), Size > 0 -> true;4 unmodified lines case ets:whereis(?TABLE) of undefined -> nil; _ -> lists:foreach( fun({Ref, {Owner, Fun}}) -> case is_process_alive(Owner) of true -> try Fun(Event) catch _:_ -> ok end; false -> ets:delete(?TABLE, Ref) end fun({Ref, {_Owner, Notify, Control}}) -> admit(Ref, Event, byte_size(Event), Notify, Control, ?CAS_ATTEMPTS) end, ets:tab2list(?TABLE)), nil end.
admit(Ref, Event, Size, Notify, Control, Attempts) -> case atomics:get(Control, 1) of 1 -> ok; 0 when Attempts == 0 -> overflow(Ref, Notify, Control); 0 -> case ets:lookup(?QUEUES, Ref) of [] -> ok; [{Ref, {Queue, Count, Bytes} = Previous}] -> case Count + 1 > ?EVENT_LIMIT orelse Bytes + Size > ?BYTE_LIMIT of true -> case replace_queue(Ref, Previous, Previous) of true -> overflow(Ref, Notify, Control); false -> admit(Ref, Event, Size, Notify, Control, Attempts - 1) end; false -> Next = {queue:in(Event, Queue), Count + 1, Bytes + Size}, case replace_queue(Ref, Previous, Next) of true -> wake(Notify, Control); false -> admit(Ref, Event, Size, Notify, Control, Attempts - 1) end end end end.
%% The separate atomic latch makes overload finite even under competing%% publishers. A stale CAS can only replace an existing row, so deletion%% prevents an in-progress admission from recreating an overflowed queue.overflow(Ref, Notify, Control) -> case atomics:compare_exchange(Control, 1, 0, 1) of ok -> ets:delete(?QUEUES, Ref), wake(Notify, Control); _ -> ok end.
wake(Notify, Control) -> case atomics:compare_exchange(Control, 2, 0, 1) of ok -> try Notify() catch _:_ -> ok end; _ -> ok end.
replace_queue(Ref, Previous, Next) -> ets:select_replace(?QUEUES, [ {{Ref, '$1'}, [{'=:=', '$1', {const, Previous}}], [{const, {Ref, Next}}]} ]) == 1.
drain(Ref) -> case ets:lookup(?TABLE, Ref) of [{Ref, {_Owner, Notify, Control}}] -> drain(Ref, Notify, Control, ?CAS_ATTEMPTS); [] -> {batch, []} end.
drain(Ref, Notify, Control, Attempts) -> case atomics:get(Control, 1) of 1 -> ets:delete(?QUEUES, Ref), overflow; 0 when Attempts == 0 -> overflow(Ref, Notify, Control), overflow; 0 -> case ets:lookup(?QUEUES, Ref) of [] -> {batch, []}; [{Ref, {Queue, _Count, _Bytes} = Previous}] -> case replace_queue(Ref, Previous, {queue:new(), 0, 0}) of false -> drain(Ref, Notify, Control, Attempts - 1); true -> case atomics:get(Control, 1) of 1 -> ets:delete(?QUEUES, Ref), overflow; 0 -> {batch, queue:to_list(Queue)} end end end end.
%% A periodic flush can overtake the wake already in the stream mailbox.%% Only consuming that wake permits rearming; otherwise its gate stays set.rearm(Ref, WakeConsumed) -> case ets:lookup(?TABLE, Ref) of [{Ref, {_Owner, Notify, Control}}] -> case WakeConsumed of true -> atomics:put(Control, 2, 0); false -> ok end, case {atomics:get(Control, 1), ets:lookup(?QUEUES, Ref)} of {1, _} -> wake(Notify, Control); {0, [{Ref, {_Queue, Count, _Bytes}}]} when Count > 0 -> wake(Notify, Control); _ -> ok end; [] -> ok end, nil.
%% Whether each session is running a turn, as its actor last announced. The%% registry reads this instead of asking a session, which may be busy.mark_running(Session, Running) -> albedo_registry:register(albedo_running, Session, Running).8 unmodified lines1223456116 unmodified linesExpand all12312412512612712712812913013113213313413513612913713813914014114214314414514614714814915046 unmodified lines//// The agents bus: one daemon-wide feed of what every session is doing, for//// the orchestrator view. It carries small events only — a session's own//// the orchestrator view. It carries small events only; a session's own//// stream stays the place for whole transcripts, tool output, and resets.
import albedo/daemon/familyimport gleam/dynamic/decode116 unmodified linesExpand all |> json.to_string, )}
/// Receive every bus event while `owner` lives. `deliver` must only send./// One subscriber's bounded queue, removed when its stream process dies.pub type Subscription
pub type Batch { Batch(events: List(String)) Overflow}
/// Publishers queue events before sending this payload-free notification.@external(erlang, "albedo_bus", "subscribe")pub fn subscribe(owner: process.Pid, deliver: fn(String) -> Nil) -> Nilpub fn subscribe(owner: process.Pid, notify: fn() -> Nil) -> Subscription
/// Take one bounded batch, keeping notification latched until sending finishes.@external(erlang, "albedo_bus", "drain")pub fn drain(subscription: Subscription) -> Batch
/// Rearm after sending. A timer can run before the pending wake was consumed.@external(erlang, "albedo_bus", "rearm")pub fn rearm(subscription: Subscription, wake_consumed: Bool) -> Nil
fn event_json( session: String, kind: String, fields: List(#(String, json.Json)),46 unmodified lines172 unmodified linesExpand all1731741751761771771781791801801811821831841337 unmodified linesExpand all15221523152415251526152715281529153015261527152815291532153015311532153515331534153715381539154015411542154315351536154515461547154815491550153715381539154015411542154315441545154615471548154915501551155215531552155315541555155615541555155615571558155915601561156215631564156515661567156815691570157115721224 unmodified linesExpand all279727982799280027902801280228032792279327942795279628042805280627982807280828092810281110 unmodified lines2822282328242825281728182819282028212822282328242825282628272828282928302831283228332834283528362837283828392826282728282829283028312832283328342835283628372838283928402841284228432844284528402841284228462847284828492850285128522853285428552856381 unmodified linesExpand all172 unmodified linesExpand all Tick Wake}
/// The agents stream: bus events arrive one by one and leave in batches./// The agents stream drains a bounded bus queue on each batching tick.type AgentStream { Flush Heard(String) Heard}
type AgentNode { AgentNode(1337 unmodified linesExpand all mist.server_sent_events( req, response.new(200), fn(self) { case process.subject_owner(self) { Ok(owner) -> bus.subscribe(owner, fn(event) { process.send(self, Heard(event)) }) Error(_) -> Nil } let assert Ok(owner) = process.subject_owner(self) let subscription = bus.subscribe(owner, fn() { process.send(self, Heard) }) process.send(self, Flush) #(self, [], 0, 50) #(self, subscription, 50, True, False) }, fn(state, message, connection) { let #(self, buffered, count, quiet) = state let #(self, subscription, quiet, initial, wake_consumed) = state case message { // Past the cap, a stalled client loses deltas rather than memory. Heard(event) -> case count < 4000 { True -> actor.continue(#(self, [event, ..buffered], count + 1, quiet)) False -> actor.continue(state) } Heard -> actor.continue(#(self, subscription, quiet, initial, True)) Flush -> { let _ = process.send_after(self, 100, Flush) case buffered, quiet >= 50 { [], False -> actor.continue(#(self, [], 0, quiet + 1)) _, _ -> { let body = event_frame( // Readiness is always an empty batch, even when events were admitted // between registering this subscription and its first flush. let batch = case initial { True -> bus.Batch([]) False -> bus.drain(subscription) } let events = case batch { bus.Batch(events) -> events bus.Overflow -> ["{\"type\":\"overflow\"}"] } let sent = case events == [] && quiet < 50 { True -> Ok(Nil) False -> mist.send_event( connection, mist.event(event_frame( string_tree.from_string("{\"events\":["), list.reverse(buffered), ) case mist.send_event(connection, mist.event(body)) { Error(_) -> actor.stop() Ok(_) -> actor.continue(#(self, [], 0, 0)) events, )), ) } case sent, batch { Error(_), _ -> actor.stop() _, bus.Overflow -> actor.stop() Ok(_), _ -> { bus.rearm(subscription, wake_consumed) let _ = process.send_after(self, 100, Flush) let quiet = case events == [] && quiet < 50 { True -> quiet + 1 False -> 0 } actor.continue(#(self, subscription, quiet, False, False)) } } } }1224 unmodified linesExpand all mist.server_sent_events( req, response.new(200), fn(self) { // The session wakes this stream per event; the tick is only a keepalive // The session coalesces wakes until a sent frame is acknowledged; the tick // is only a keepalive // and the way a dropped connection is noticed while nothing is streaming. case process.subject_owner(self) { Ok(owner) -> session.watch(worker, owner, fn() { process.send(self, Wake) }) Error(_) -> Nil } let assert Ok(owner) = process.subject_owner(self) session.watch(worker, owner, fn() { process.send(self, Wake) }) process.send(self, Tick) #(self, after) #(self, after, owner) }, fn(state, message, connection) { // A closing daemon stops session workers while clients are still attached, // so a dead worker ends this stream instead of failing a call into it.10 unmodified lines |> mist.event_name("error") let _ = mist.send_event(connection, failure) actor.stop() } Ok(page) -> case message, page.events { Wake, [] -> actor.continue(state) _, _ -> { let events = event_frame( string_tree.from_strings([ "{\"generation\":", json.to_string(json.string(page.cursor.generation)), ",\"cursor\":", int.to_string(page.cursor.sequence), ",\"events\":[", ]), page.events, ) case mist.send_event(connection, mist.event(events)) { Error(_) -> actor.stop() Ok(_) -> { case message { Tick -> { let _ = process.send_after(state.0, 1000, Tick) Nil } Ok(page) -> { let events = event_frame( string_tree.from_strings([ "{\"generation\":", json.to_string(json.string(page.cursor.generation)), ",\"cursor\":", int.to_string(page.cursor.sequence), ",\"events\":[", ]), page.events, ) case mist.send_event(connection, mist.event(events)) { Error(_) -> actor.stop() Ok(_) -> { session.consumed(worker, state.2, page.cursor, message == Wake) case message { Tick -> { let _ = process.send_after(state.0, 1000, Tick) Nil Wake -> Nil } actor.continue(#(state.0, Some(page.cursor))) } Wake -> Nil } actor.continue(#(state.0, Some(page.cursor), state.2)) } } } } }, )}381 unmodified linesExpand all139 unmodified linesExpand all140141142143144145146147148237 unmodified linesExpand all386387388389390391392393394395396397398399400401402403404314 unmodified linesExpand all719720721722711712713714715716723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757718719758759760761762763764271 unmodified linesExpand all103610371038103999910401041104210431044104510461047104810491050105111 unmodified lines1063106410651066106710681069107010711325 unmodified linesExpand all139 unmodified linesExpand all ChangeEffort(String, Subject(Result(json.Json, String))) Status(Subject(String)) Read(Option(Cursor), Option(Int), Subject(Result(Page, String))) Watch(process.Pid, fn() -> Nil) Consumed(process.Pid, Cursor, Bool) Publish(String, String, Subject(Bool)) Commit( String, List(types.Input),237 unmodified linesExpand allpub fn watch(session: Session, owner: process.Pid, notify: fn() -> Nil) -> Nil { process.send(session, Watch(owner, notify))}
/// A successful frame consumed this cursor. Timers can overtake a queued wake,/// so only handling that wake rearms its notification.pub fn consumed( session: Session, owner: process.Pid, cursor: Cursor, wake_consumed: Bool,) -> Nil { process.send(session, Consumed(owner, cursor, wake_consumed))}
/// Events after `after`. A reset (a new or lagging client) replays the/// transcript: whole, or with `tail` only its newest rows, which the client/// pages back from with `history.rendered`.pub fn read(314 unmodified linesExpand all ), ]) |> json.to_string, ) Watch(owner, notify) -> actor.continue( session_state.State(..state, watchers: [ #(owner, notify), ..list.filter(state.watchers, fn(watcher) { process.is_alive(watcher.0) && watcher.0 != owner Watch(owner, notify) -> { let watchers = case list.any(state.watchers, fn(watcher) { watcher.owner == owner }) { True -> state.watchers False -> { let _ = process.monitor(owner) [session_state.Watcher(owner, notify, False), ..state.watchers] } } actor.continue(session_state.State(..state, watchers: watchers)) } Consumed(owner, cursor, wake_consumed) -> { let watchers = case cursor.generation == state.generation && cursor.sequence >= 0 && cursor.sequence <= state.sequence { False -> state.watchers True -> list.map(state.watchers, fn(watcher) { case watcher.owner == owner { False -> watcher True -> { let notified = watcher.notified && !wake_consumed case !notified && cursor.sequence < state.sequence { True -> { watcher.notify() session_state.Watcher(..watcher, notified: True) } False -> session_state.Watcher(..watcher, notified: notified) } } } }) ]), ) } actor.continue(session_state.State(..state, watchers: watchers)) } Read(after, tail, reply) -> { let cursor = Cursor(state.generation, state.sequence) let replay = case after { Some(after) if after.generation == state.generation ->271 unmodified linesExpand all case turn.owner(state.activity, id) { Some(run) -> finish_run(state, run, outcome) _ -> actor.continue(state) } Down(process.ProcessDown(_, pid, _)) -> Down(process.ProcessDown(_, pid, _)) -> { let state = session_state.State( ..state, watchers: list.filter(state.watchers, fn(watcher) { watcher.owner != pid }), ) case turn.running(state.activity) { Some(run) if run.pid == pid -> case run.work { turn.Background(reply) ->11 unmodified lines ) } _ -> actor.continue(state) } } Down(_) -> actor.continue(state) Idle(reply) -> { let #(kernel, jobs) = case state.kernel { Some(kernel) -> #(1325 unmodified linesExpand all11 unmodified lines1213141516171819202122236 unmodified lines3031323330343536373816 unmodified lines555657585556596061626364656667686970715 unmodified lines11 unmodified linesimport gleam/erlang/process.{type Subject}import gleam/listimport gleam/option.{type Option}
pub type Watcher { Watcher(owner: process.Pid, notify: fn() -> Nil, notified: Bool)}
pub type State(message) { State( info: conversation.Info, host: runtime.Runtime,6 unmodified lines steering: List(Submission), active_submissions: List(Submission), sequence: Int, events: event_buffer.Buffer, watchers: List(#(process.Pid, fn() -> Nil)), watchers: List(Watcher), notice: Option(String), context: context_snapshot.Snapshot, pin: loop.Pin, /// Inputs evicted by the last prepared projection; the next live reload16 unmodified linespub fn emit(state: State(message), event: String) -> State(message) { let seq = state.sequence + 1 let events = event_buffer.push(state.events, seq, event) let watchers = list.filter(state.watchers, fn(watcher) { process.is_alive(watcher.0) }) list.each(watchers, fn(watcher) { watcher.1() }) list.map(state.watchers, fn(watcher) { case watcher.notified { True -> watcher False -> { watcher.notify() Watcher(..watcher, notified: True) } } }) bus.activity(state.info.id, event) State(..state, sequence: seq, events: events, watchers: watchers)}
5 unmodified lines