Feat/events socket#352
Conversation
Adds the Event enum and RemovedReason for the push-event socket. Wire format is JSONL: snapshot reuses shpool_protocol::Session so the schema matches `shpool list --json`; deltas are flat objects tagged with `type`.
The daemon binds a sibling Unix socket (events.socket) next to the main shpool socket and accepts long-lived subscribers. Each new subscriber receives a snapshot of the session table as its first message, built under the shells lock so subsequent deltas (published in a follow-up change) cannot race the registration. Per-subscriber writer threads with bounded channels and a write timeout isolate slow or stuck consumers from the daemon's hot path; subscribers that fall behind are dropped and re-sync by reconnecting. Extracts collect_sessions from handle_list so the snapshot and the existing `shpool list --json` output share one schema-producing path.
Emits session.created, session.attached, session.detached, and
session.removed (with reason exited|killed) at the seven mutation
sites that change the session table:
* select_shell_desc create path: created + attached
* select_shell_desc reattach path: attached
* handle_attach client-disconnect path: detached
* handle_attach shell-exit path: removed{exited}
* handle_detach: detached
* handle_kill: removed{killed}
* ttl_reaper expiry: removed{killed}
Each publish runs inside the same shells-lock scope as its mutation,
so wire-order matches causal-order for any subscriber. Reaping is
surfaced as `killed` for now; a dedicated reason can be added later
if a use case appears.
Connects to the daemon's events socket and prints each JSON line to stdout, flushing per line so the stream is pipeline-friendly: shpool events | jq 'select(.type == "session.removed")' The first line is a snapshot of the current session table; subsequent lines are deltas. Reconnect to force a fresh snapshot.
Three end-to-end tests, exercising the full daemon → events socket →
JSON wire path:
* snapshot_then_lifecycle: snapshot, then session.created /
.attached / .detached / .removed{killed} as a session is created,
detached (via background mode), and killed.
* snapshot_includes_existing_sessions: a subscriber that connects
after a session already exists receives that session in the
snapshot.
* multiple_subscribers_each_get_independent_streams: two concurrent
subscribers both receive the snapshot and full delta sequence.
- Gate the SessionRemoved{Exited} publish in handle_attach's shell-exit
branch on shells.remove() actually returning Some, so a concurrent
kill or reaper that already removed the entry doesn't produce a
duplicate removal event.
- Drop the eager SessionDetached publish in handle_detach. The
bidi-loop unwind path in handle_attach already publishes the matching
event with its own timestamp; emitting it twice was observable to
subscribers. Keep the eager last_disconnected_at write so concurrent
list() callers still see fresh state immediately.
- In select_shell_desc, defer the reattach SessionAttached publish past
the is_finished() check so it isn't emitted for a session about to be
implicitly clobbered by the create path. Have the create path publish
SessionRemoved{Exited} when it overwrites an existing entry, so the
replacement is explicit on the wire.
- Thread the events socket path through signals::Handler so signal
exits clean it up alongside the main socket. Switch to a Vec<PathBuf>
to handle both. Tolerate NotFound on cleanup since the events socket
may not have been bound yet when a signal arrives.
- Drop the unnecessary `pub` on Server.events_bus.
- Apply nightly rustfmt.
Three new integration tests:
- explicit_detach_publishes_one_event: pins the no-duplicate-detached
invariant by using a kill as a known-next-event fence; a duplicate
detached would surface as the next read instead of session.removed.
- signal_exit_unlinks_sockets: SIGTERM the daemon and assert both
socket files are gone.
- reattach_emits_attached_only: regression guard for the reattach path.
|
Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA). View this failed invocation of the CLA check for more information. For the most up to date status, view the checks section at the bottom of the pull request. |
This would have helped for debugging shell-pool#335. Let's add it now, better late than never.
|
I really appreciate your enthusiam, but in the future, let's try to do more up-front design before jumping into coding. #346 (comment) explains some more of my thoughts on the matter. Rather than jumping straight to code, can you first post the design you intend to code to, with a particular focus on the contract and API that we'll offer. Basically, I think we should work out and agree on what the developer-facing documentation will look like (doesn't have to be polished, but we should have a precise spec that a developer for a tui could code against and agree on it). Normally and issue would be the place to do that, but we can just do it in this thread and then you can update the PR contents once we've settled on something. Some feedback on what I see in the PR description:
|
|
Re less rich events, how about just the One goal of the current design was to make it so that an external program can, solely by watching the events socket, maintain a consistent view of the shpool state. This required the snapshot-upon-subscription event (for synchronization) and some metadata (like So, if we simplify the events, the events stream is no longer sufficient, so it might as well be simplified further to no longer fire any -upon-subscription event (which also means events.rs doesn't need to Re your Another benefit of |
|
Yeah, keeping an Event enum seems like a good idea. I think dropping the snapshot and the extra fields besides type and name seem pretty good. In terms of hiding the socket location, I still think we should document the protocol and socket location. I do think that heavier duty applications should probably be just directly dialing the socket rather than forking a sub-proc, so we should be intentional about making that part of the public API (people will just do it anyway and we'll be on the hook for compatibility so we might as well be intentional). I think the way you have it with as sidecar events.socket file. |
|
Maybe we could even get away with dropping name as well. Why do you need to be able to track which specific session changed? |
This patch improves lock contention on the global sessions lock by releasing it when sending resize and detach messages to the shell->client thread. Previously, we could lock up the global table for up to SESSION_MSG_TIMEOUT, which is quite bad.
This patch fixes an issue where we were not exiting on EOF correctly when scanning for the shell startup sentinel.
Co-authored-by: release-plz-for-shpool[bot] <175248994+release-plz-for-shpool[bot]@users.noreply.github.com>
|
I don't need the I think the external documentation would just cover the event types and serde-derived JSONL format of the stream, accessible via Would the documentation live in a top-level EVENTS.md that README.md points to? |
|
Yeah, I think that would be a good place to put the documentation. We probably don't want a section inline in the README so a seperate .md file is a good way to go. That all sounds like a good plan. Do you want to update this PR with that plan in mind? Once you are done, just let me know and I'll review. |
Each event now serializes as `{"type":"session.<x>"}` with no other
fields. To learn what the event refers to, subscribers follow up with
`shpool list` (or send `ConnectHeader::List` over the main socket).
- Drop the welcome `snapshot` event; subscribers do their own bootstrap
list call after connecting.
- Drop `name`, timestamps, and `reason` from the lifecycle events.
- Remove `RemovedReason`; reaped/killed/exited share `session.removed`.
- Inline `collect_sessions` back into `handle_list` (the snapshot was
its only other consumer) and drop the now-unused `unix_ms` helper.
Cover the sibling-socket transport, the four event types, the JSONL wire format, the `shpool events` CLI helper plus direct-socket use for heavier-duty consumers, the ordering guarantee (publish under the session-table lock so wire-order matches causal-order), and the slow-subscriber drop policy.
|
I've drafted EVENTS.md and stripped out the extra fields. It looks like the CLA check is unhappy about the |
ethanpailes
left a comment
ethanpailes
left a comment
There was a problem hiding this comment.
EVENTS.md looks pretty reasonable, just a few bits of feedback. Let me know when the rest of the change is ready for review and I'll give it a full read through.
| Each subscriber has a bounded outbound queue. A subscriber that falls too far | ||
| behind is dropped by the daemon (in which case the subscriber can always reconnect). | ||
| There is no replay, so events that fired while a subscriber was disconnected are | ||
| lost. |
There was a problem hiding this comment.
Perhaps we should document a knob in the config.toml that tunes how large this queue is.
There was a problem hiding this comment.
Do you want the config knob added in this PR?
The queues currently use std::sync::mpsc::sync_channel here, pre-allocating all SUBSCRIBER_QUEUE_DEPTH slots.
Would you want to use something slightly slower that allocates on-demand and has something like shrink_to_fit to return memory after a spike? I'm just imagining if someone sets their queue depth to 1 billion not expecting it to pre-allocate for each subscriber. If it instead allocates on-demand you could maybe set the queue depth to something large like 100K by default anyway, at which point maybe it wouldn't be worth offering a config knob.
There was a problem hiding this comment.
Hmm, that's a good point. Honestly I should probably suppress the urge to throw a knob for everything. Who is the target demo that would actually tune this knob? Probably a tool developer who would tell their users to configure shpool specially, but that seems pretty unlikely. If it is a real problem they can always open an issue and we can expose the knob then.
There was a problem hiding this comment.
Orthogonal to the knob, would you want a dynamically allocated data structure to replace sync_channel anyway? If a user goes wild with sessions, e.g. a bunch of programmatically generated sessions with different @{var} combinations, then Mutex<VecDeque>/shrink_to_fit might scale better.
Actually, since the events are uninformative by design, subscribers probably won't care which events are in the queue so much as if an event is in the queue. So, what about sync_channel(1) (a single slot)? Option would work too but sync_channel makes it easy to increase to 8 or 16 if events become more informative later on.
This patch bumps the shpool_pty crate version to pick up some soundness fixes. I noticed that it's resource management was unsound when I realized that it both allowed cloning and did not use reference counting, so I fixed that. I think the only reason this wasn't burning us before was that the lifetime for the Fork object was outliving all the Master and Slave objects pulled off of it (either that or we were just not cleaning up resources at all).
This patch tweaks the AI policy again.
|
@GeoffChurch are you ready for me to review the main PR body or are you still chugging? |
|
I think the only thing left is a question I had about the config knob: |
|
Sounds good, I'll give this a review soon. |
|
|
||
| /// Broadcast `event` to all current subscribers. Subscribers whose | ||
| /// queues are full or whose receivers have hung up are dropped. | ||
| pub fn publish(&self, event: &Event) { |
There was a problem hiding this comment.
This looks like more work than I would like to see in the publish routine. I think we shuld architect this such that the publish routine does as little wor as possible. Here it is looping over a vec while hoding a lock. I would rather we just do a send on a crossbeam channel and then have a sink thread that picks up the even and handles it. The sink thread can just have a big poll loop that hecks if the subscribers who still need messages arewriteable. We kick any subscriber who has a full buffer and is not writeable. that would also allow us to void having a thread per subscriber since we would b essentially doing asysnc io directly.
| for stream in listener.incoming() { | ||
| match stream { | ||
| Ok(stream) => { | ||
| if let Err(e) = on_accept(stream) { |
There was a problem hiding this comment.
We should probably fork a thread befor each connection so that multiple people can dial in in parallel.
| Ok(()) | ||
| } | ||
|
|
||
| fn run_writer(mut stream: UnixStream, receiver: Receiver<Arc<str>>) { |
There was a problem hiding this comment.
As mentioned above, to shift work away from the evnt emitter code, I think we should replace these wit one big sink that has a big poll loop.
|
Sorry for all the typos. I'm on a slow connection and it seems like github round-trips edits so it is really painful to type in the review boxes. |
GeoffChurch
force-pushed
the
feat/events-socket
branch
3 times, most recently
from
f2a7338 to
c030359
Compare
- Add .context() on the signal-handler spawn and events-listener startup paths in daemon::run. - Move the ttl-reaper thread::spawn's capture-clones (shells, events_bus) inside a block expression. The cloned Arcs shadow the outer names rather than introducing parallel shells_tab / reaper_bus bindings in the function scope. - Reword the comment on the shells-remove + SessionRemoved publish to explain why the publish is gated on is_some(): a concurrent kill or reaper may have removed the entry (and published) while we were waiting for the lock. - Fold three single-use SystemTime::now() locals back inline. - Have start_listener take Arc<EventBus> directly instead of a closure that registers + spawns a writer. EventBus::register and spawn_writer drop their pub modifiers now that the only call site is within events.rs. - Restore the let-binding type annotation on the collected sessions Result in handle_list. - Document JSONL newline handling in EVENTS.md.
- bus_publish_with_many_subscribers_is_not_quadratic: bound a single publish to 10K subscribers under 50 ms (~4 ms on a 2.1 GHz CPU; 50 ms absorbs CI tail latency while still catching quadratic regressions, which would be on the order of seconds at N=10K). - bus_drops_slow_subscriber_on_overflow_without_affecting_fast: a fast subscriber keeps receiving every event while a slow one's queue fills and is eventually dropped on overflow. Asserts SUBSCRIBER_QUEUE_DEPTH stays small since the test scales with it. - accept_loop_registers_concurrent_subscribers: dial 20 connections in parallel and confirm all are registered and each receives a published event. - bus_concurrent_publish_under_outer_lock_delivers_all_events: publish from multiple threads while each holds an outer mutex; all events arrive in order.
GeoffChurch
force-pushed
the
feat/events-socket
branch
from
e33e751 to
5250954
Compare
2 participants
Issue Link
#346 (comment)
AI Policy Ack
ack -- pair-programmed with Claude; every line was human-reviewed.
Description
Adds an events socket where the shpool daemon can push events to subscribers. This lets external programs like TUIs stay updated with the shpool state w/o needing to poll. The new events socket is located at
<runtime_dir>/shpool/events.socket, alongside the main socket.The format is JSONL (each event is a one-line JSON object). Events look like:
This PR also (tentatively) includes
shpool events(seefn subscribe_to_stdout), which is a QoL subcommand that connects to the events socket, prints each event as its received, and flushes after each print. This alleviates some boilerplate from downstream users and makes it easy to pipe the output to other programs.Each subscription spawns a writer thread with its own internal buffer and timeout. This is done to avoid blocking on slow event consumers.