feat(pelican): stream file events over SSE at GET /pelican/subscribe

[rbardaji]

Closes #262, alongside /pelican/read (#265).

Why

The event server notifies subscribers whenever an object appears in a Pelican namespace, but that stream existed only in the ndp-ep client library. The API had no way to expose it, so an Endpoint could not offer subscriptions to its own callers.

The routes

GET /pelican/subscribe?event_source=osdf/vdc/public/data

: subscribed

event: file
data: {"name":"a.csv","url":"pelican://...","size":42,"mod_time":"2026-08-30T12:00:00Z",...}

: keepalive

Each event's url goes straight to /pelican/read for the contents, or /pelican/download for the file. That pipeline — subscribe, then read what arrived — is the reason the route exists.

GET /pelican/subscriptions

Reports the upstream subscriptions the Endpoint currently holds: connection state, listener count, and how many events were dropped for slow listeners.

Design decisions worth reviewing

One upstream subscription per event source, fanned out to every listener. The event server requires a unique client-id per subscriber: two connections sharing one compete for the same events instead of both receiving them. A subscription per SSE caller would therefore either split the stream between callers or leave an orphaned client id registered upstream on every connection. Instead the Endpoint subscribes once per event source under its own stable id and distributes each event to all its listeners. The upstream opens when the first listener arrives and closes when the last leaves, so an idle Endpoint holds no connection to the event server.

The STOMP wire format is implemented here rather than taken from the client library. Depending on ndp-ep would make the API depend on its own client SDK and pull in pelicanfs, which pins Python 3.11. The cost, stated in the CHANGELOG: the protocol now lives in two repositories and has to be kept in step by hand.

Per-listener buffers are bounded, unlike the client library's deliberately unbounded queue. A caller that stops reading must not be able to grow the Endpoint's memory without limit, so a full buffer drops its oldest event and counts it in /pelican/subscriptions.

The idle timeout lives inside the generator, not around it. Cancelling an __anext__ from outside would unwind the generator and tear the subscription down on every quiet interval, so listen() times its own wait out and yields None for the route to turn into a : keepalive. There is a test pinning this.

Also: X-Accel-Buffering: no, or nginx buffers the stream and holds events back until the buffer fills, defeating the point.

Verification

59 new cases in tests/test_pelican_events.py: STOMP header escaping (including a round trip over all four escapes), frame encode/parse, event parsing and its rejection cases, redelivery identity, the environment resolution and every one of its fallbacks, the fan-out itself (every listener gets every event; a slow listener drops its oldest; a duplicate is suppressed but still acknowledged, since an unacknowledged message is redelivered forever), and the routes.

Against the real app: no token gives 401, an unconfigured Endpoint gives 503 naming the missing setting, /pelican/subscriptions answers 200, and both routes appear in the OpenAPI schema.

Local gate: black and flake8 clean, 1349 tests pass. The three failures in tests/repositories/test_catalog_settings.py and tests/test_publi_env.py predate this branch — they reproduce on a clean main and come from a local .env being read by the settings tests, which does not exist in CI.

Not verified: the live connection to the event server. There are no credentials to hand, so the STOMP path is exercised against simulated frames, not against stomp-server.chtcdev.chtc.io. Reviewing that end of it needs an account on the event server.

Backwards compatibility

Purely additive. Both routes sit behind the authorization added in #261 and are only mounted when PELICAN_ENABLED is set.

Subscriptions need PELICAN_EVENT_CLIENT_ID, or an AFFINITIES_EP_UUID to derive it from; without either, /pelican/subscribe answers 503 with the reason, so an Endpoint that never configures it is unaffected. The remaining settings (PELICAN_EVENT_SERVER_URL, PELICAN_EVENT_USERNAME, PELICAN_EVENT_PASSWORD, PELICAN_EVENT_VIRTUAL_HOST, PELICAN_EVENT_HEARTBEAT_MS) are optional and documented in example.env and docs/configuration.md.

websockets is a new dependency, required by the event server's transport.