One Way Pattern: canonical SSE collection feeds; v1 ui.grid.* frames removed - #76
Merged
Conversation
…e port Re-seat the SSE byte-writing path on the new core contract without moving the transport mechanism (poll loop, deliver() ladder, Redis durability, sessionWorkerTable/deliverTable, disconnect, coroutine mgmt all untouched). - SwooleSseTransport: Swoole adapter implementing core SseTransportInterface; the one instanceof Swoole\Http\Response cast (mirrors SwooleResponseEmitter). - AsyncResourceSseServer: writeSse/writeSseComment now go through the port; composeSseFrame() -> buildFrame(): SseFrame. resolveSseEventName() and the UiSseEventType allow-list STAY here (the chokepoint moves to this consumer boundary, not core) and now produce a validated core SseFrame. - TypedSseFrameTest: reflection target retargeted to buildFrame()->toWire(); wire-format assertions unchanged — byte-identical (14 tests, 58 assertions). No behaviour change; full SSE Async suite green (83 tests). PHPStan level max clean; module_structure 0 violations. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ad /__semitexa_sse Honestly declare the two ssr SSE streams' existing gate models for the reformulated boot guard, and retire the dead /__semitexa_sse path. - /sse (SseEndpointPayload): sseGateModel: BearerSession. - /__semitexa_kiss (SseKissPayload): add transport:Sse + sseGateModel: BearerSession — closes the coverage hole (a live stream that never declared its transport). Shares the handleSse code path with /sse, so the identical gate model. - /__semitexa_sse: dead code removed. Split the handle() OR so /sse and /__semitexa_kiss keep working; delete the orphaned async-resource-manager.js client (never instantiated — nothing emits [data-async-resource]) and its assets.json override. No residual /__semitexa_sse reference remains. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
/sse was a byte-identical duplicate alias of /__semitexa_kiss (same handleSse code path). With UI streaming unified on kiss: - Delete SseEndpointHandler + SseEndpointPayload (drops the /sse route + its BearerSession declaration). - Collapse the AsyncResourceSseServer::handle intercept from ($path==='/sse' || '/__semitexa_kiss') to kiss-only. - Reword the SseKissPayload + handleSse comments to drop the /sse reference. Shared kiss machinery untouched (handleSse, AsyncResourceSseServer, SseKissPayload/Handler, message publisher, listener, UiSseEventType). AsyncResourceSseServerTest 69/69; AllRoutes/AuthenticatedRoutes smoke green; ai:verify pass; module_structure 0. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The two non-owner-request-scoped writers on AsyncResourceSseServer — deliverToUser() (all of one user's sessions) and deliverToAuthenticatedUsers() (system-wide) — plus the DI-bound AsyncResourceSseMessagePublisher::publishToUser() wrapper do zero content-vs-recipient authorization: they merely loop owner-scoped deliver() over a recipient list. They are latent (zero callers) today, so private content (e.g. a grid UiComponentStateMessage row bag) could ride a fan-out to non-entitled sessions the moment a caller is wired. Fence them shut now, while fencing is free: each is marked @internal and throws FanOutNotYetGatedException before any deliver()/socket write, so no frame can leak even partially. Original bodies are preserved as Track-R restoration comments with a per-recipient-check insertion point. This is a barrier, not the filter. Owner-scoped deliver()/publish() and live kiss delivery are untouched. Track R replaces the throw with the real per-recipient entitlement-gated implementation. Verify: AsyncResourceSseServerTest 69/69; ai:verify 7/7; module_structure 0; discovery boots (kiss route + DI contract intact). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Build the store that later Track R bricks consume — R5 inserts on connect, R3 queries the reverse index on a push, R4 reads the DTO. Three tiers with deliberately distinct representations and lifetimes; the separation is a security boundary (design §C.5/§C.6, §B.1). Tier 1 — SubscriptionTable: cross-worker, SERIALIZED Swoole\Table, one row per subscription keyed by streaming_id. String-only schema (streaming_id, session_id, tenant_id, scope_keys, tenant_blob) — no DTO/object column, so a live identity-bearing object cannot enter this tier by construction. The tenant_blob is the opaque serialized tenant context R2 deliberately did not hold worker-local; it belongs here because this row is read cross-worker. Tier 2 — SubscriptionDtoRegistry: worker-static, NEVER serialized map streaming_id => live ReRunContext (R2's worker-local re-run state). The only tier holding the live object; safe because dispatch_mode 2 pins the fd to the owning worker for the stream's life. Tier 3 — SubscriberIndexInterface + ScanningSubscriberIndex: reverse index find(tenant, scopeKey) -> list<SubscriberRef> via an O(rows) scan, with the keyed-Table seam (design §C.5) noted so the impl swaps without touching callers. Tenant is part of the lookup key (security boundary: a push in tenant A never resolves tenant B subscribers on a same-named scope). scope keys are P1 resourceKeys (default = table name), matching what a P2 ResourceChangedEvent carries. Proven in isolation over a real Swoole\Table (6 tests / 45 assertions): the subscription round-trip, the worker-static DTO round-trip, the tier-separation invariant (no live object in the cross-worker table; live state only in the registry), the reverse-index seam (swappable behind the interface), and the scope-key/P1 linkage against a real ResourceChangedEvent. No connect, subscriber, loop, or re-run code (R5/R3/R4/R2 scope). ai:verify pass — module_structure 0, phpstan_di 0; ssr Unit/Async 89/89. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Build the publisher side of the push path (publisher-side ONLY): an
#[AsEventListener] on the orm P2 ResourceChangedEvent that resolves the
tenant from ambient TenantContext and PUBLISHes a DATA-LESS scope signal
to ui.invalidate.{tenant}.{scopeKey} — the exact channel R1's reverse
index find(tenant, scopeKey) resolves against, with scopeKey = the P1
resourceKey the event carries (producer/subscriber agree by construction).
Pieces (semitexa-ssr only):
- ResourceInvalidationPublisher: the listener; references the orm event by
FQCN string + reads only $event->resourceKey (no orm Composer dep / import
in src; module_structure + phpstan_di stay 0). Operation deliberately not
published — data-less; the channel name carries the full routing key.
- ScopeInvalidationBusInterface + SseRedisScopeInvalidationBus: narrow
publisher-side seam (publish(channel) only, no subscribe), reusing the
size-1 SSE pool via AsyncResourceSseServer::publishScopeInvalidation()
(PUBLISH is non-blocking request/reply → pool is safe; dedicated is
subscriber-only, design §C.3).
- PinResourceInvalidationPublisherSyncListener: STRUCTURAL synchrony pin
(GATE-1 §T5). Reflects the declared #[AsEventListener] execution mode and
boot-fails (ConfigurationException) unless Sync — enforced at worker boot
AND in the constructor. An Async/Queued drift would lose the ambient
tenant and mis-scope the channel → cross-tenant invalidation leak; the
pin makes that a hard failure, not a silent leak. Proven by deliberate-red.
No subscriber, no SUBSCRIBE, no reverse-index lookup, no re-run/loop/connect,
no dispatcher-into-OrmManager wiring (manual dispatch in tests). ai:verify
pass 10/10 (module_structure 0, phpstan_di 0, AsyncResourceSseServerTest
69/69); ssr Async suite OK (98 tests).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The cross-instance push RECEIVER: a per-worker coroutine that SUBSCRIBEs to the
ui.invalidate.{tenant}.{scopeKey} channels P3 publishes to, resolves local
subscribers via R1's reverse index find(tenant, scopeKey), and routes ONE
coalesced {__ctrl:rerun} control to each session-addressed queue (the existing
deliver() path). Subscriber + routing ONLY — it does NOT execute the re-run (R4)
and does NOT drive the connect lifecycle (R5).
Three load-bearing invariants:
- DEDICATED connection (HARD, design §C.3): RedisSubscribeConnectionFactory only
ever new Client(...) — never the size-1 SSE pool; proven structurally (distinct
clients per call + comment-stripped code has no getRedisPool/RedisConnectionPool).
- IDEMPOTENT: RerunCoalescer collapses duplicate signals to one pending re-run per
stream via an atomic Swoole\Table incr (cross-worker safe, no lock); clearPending
is the R4 seam.
- TENANT ISOLATION: find filters on both channel dimensions — a t1 signal never
resolves a t2 subscriber on a same-named scope.
Lifecycle (C2): desiredChannels()/channelDiff() expose subscribe-on-first /
unsubscribe-on-last off R1's current store state — the seam R5 drives.
Proven in isolation by manual PUBLISH (no live ORM write, no live Redis, no
running loop): publish→one control, tenant isolation, dedicated-connection
(structural), N-rapid-signals→one re-run, lifecycle channel set, and the no-rerun
fence. ai:verify pass (module_structure 0, phpstan_di 0); ssr Async 109/109 green
(was 98; +11). Not pushed.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Populate/depopulate the three-tier subscription store R3/R4 consume.
C1 onConnect: insert the tier-1 SubscriptionRecord into the cross-worker
SubscriptionTable (visible to all workers) and store the live ReRunContext
(R2) in the worker-local SubscriptionDtoRegistry on the owning worker
(never serialized, never another worker); subscribe-on-first by driving
R3's channelDiff seam.
C2: CreateTrackRTablesListener creates both cross-worker shared surfaces
(tier-1 SubscriptionTable + RerunCoalescer) once, pre-fork (PreStart), via
their single-schema-site create() factories — the coalescer lifecycle R3
deferred to R5.
C3 onDisconnect: remove the tier-1 row, remove the tier-2 ReRunContext,
clearPending on the coalescer, and unsubscribe-on-last — no zombie left in
any tier.
ChannelSubscriptionControllerInterface is the apply seam, so R5 is proven
on a SYNTHETIC connect (capturing double; real Swoole\Table; no live
HTTP/Redis/coroutine). R5 neither catches {__ctrl:rerun} nor runs the
re-run (R4) and is not the live kiss endpoint (R8).
Tests: 8/57 new; ssr Unit/Async 117/317 green (AsyncResourceSseServerTest
green). ai:verify pass: module_structure 0, phpstan_di 0.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
… cycle)
The kiss poll/drain loop now distinguishes a {__ctrl:rerun} control marker
from a data frame at all three session-queue drain points (same-worker
$queues, cross-worker $deliverTable, and the Redis X→W seam). On a control
it resolves the worker-local ReRunContext (R1 tier-2) by streaming_id, runs
R2's ReRunner auth-first, writes the freshly re-queried frame OR closes the
stream on TERMINATE (no data frame on lost access), then clears the R3
coalescer mark so the next mutation re-arms — the bounded coalescing window.
A control is a signal, never bytes: it never reaches the client. A missing
tier-2 context (non-owner worker drained it, or the stream was torn down) is
a safe no-op — no crash, no re-run, no frame. The re-runner + coalescer are
worker-static seams (setReRunner/setRerunCoalescer), null until R8/the
dispatcher brick lights them up, so R4 is inert-but-harmless until wired.
Proven against a REAL R5-populated store (ConnectCoordinator::onConnect),
with a fake R2 re-runner and a capturing transport: fresh frame (re-queried,
not stale), TERMINATE close, clearPending re-arm, cross-worker correctness
incl. the missing-context edge, and the data-less-delete edge (empty/"gone"
frame, no stale data). Push→re-run cycle is closed.
No dispatcher wiring, no live endpoint (R8), no SseGateModel/fence/tag touch.
ai:verify pass 6/0 (module_structure 0, phpstan_di 0); Unit/Async 127/356;
AsyncResourceSseServerTest 69/75 (non-control data path unchanged).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The SSE serve dispatch in AsyncResourceSseServer::handle() was hardcoded to the single '/__semitexa_kiss' path, so an own-route #[AsProtectedPayload( transport: Sse)] endpoint could not be served. Key the intercept on the matched route's declared transport instead of its path. - AsyncResourceSseServer: replace the `$path === '/__semitexa_kiss'` check with shouldServeAsSse(), an O(1) lookup against a per-worker served-path set; add setSseServedPaths() (index shape only — the transport filter lives in the listener). - WireSseServedPathsListener (WorkerStartAfterContainer, requiresContainer): reads RouteRegistry and registers every route whose transport is TransportType::Sse. /__semitexa_kiss declares transport: Sse, so it lands in the set and is served by the same generalized path — KISS is unchanged. Purely additive plumbing: enables own-route SSE serving but switches nothing on. No grid route (R8b), no cutover (R8c), no consumer-half launch. The boot guard (AttributeDiscovery::assertSseGateCoherence) is untouched. Verified: kiss serves text/event-stream live after a clean restart; unit proofs that the dispatch keys on transport (collectSsePaths + shouldServeAsSse, synthetic non-kiss Sse path served, non-Sse path not); ai:verify pass, module_structure 0, lint:di OK (259 classes), ssr Async/Server 132/132. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Convert the own-route SSE serve from a one-shot frame to a HELD-OPEN stream serviced by the same drain loop kiss uses, and wire the live consumer-half so a real mutation delivers a fresh re-queried frame on the SAME held-open fd. - AsyncResourceSseServer: extract runHeldOpenLoop() (kiss byte-identical) and add serveResourceStream() — held-open own-route serve that writes the initial frame, runs R5 onConnect/onDisconnect around the loop, and is serviced by R4. Cure the re-run socket-reentrancy snag with a coroutine-local re-run guard (isReRunInProgress) so the re-invoked own-route handler degrades to a JSON body instead of grabbing the live fd. Add setConnectCoordinator(). - LivePubSubChannelController: the live ChannelSubscriptionControllerInterface (launches R3's subscribe loop on subscribe-on-first; single-loop model logged). - WireTrackRConsumerListener (WorkerStartAfterContainer): first production wiring of R1-R5 — builds the R1 index, R3 subscriber, R5 coordinator and R2 RouteReRunner; arms R4 (setReRunner/setRerunCoalescer). Fail-soft + logged. - UiSseEventType: add ui.grid.data / ui.grid.error to the typed-event allow-list so the initial frame and every re-run frame travel the one chokepoint. Proven: HeldOpenGridStreamTest drives the full real consumer-half chain (publish -> R3 -> deliver -> R4 -> R2 -> fresh frame on the same captured fd; N mutations -> N frames), TERMINATE closes with no data frame, onConnect/onDisconnect, and the re-run guard. ai:verify pass (module_structure 0, phpstan_di 0); ssr Async+Server 136/136; kiss held-open stream unregressed live; grid guest -> 401 live. Live authenticated end-to-end is environment-blocked (semitexa.pl wires no User auth) — integration-proven instead. Old path/frontend untouched (R8c-2). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
… idle + drops
The held-open live-update silently stopped after ~60s: the R3 subscriber's
dedicated Predis pubSubLoop inherited PHP's default_socket_timeout, so an idle
SUBSCRIBE threw ConnectionException ("Error while reading line from the server")
and the loop logged + returned with NO reconnect — and was only ever relaunched
on the next connect's channel-diff, so a drop while idle-but-subscribed left the
worker permanently deaf to invalidations.
Two fixes (ssr-only, transport plumbing — no auth / anti-poisoning touched):
- RedisSubscribeConnectionFactory: set read_write_timeout => -1 on the dedicated
connection. A parked SUBSCRIBE must never idle-timeout.
- ResourceInvalidationSubscriber::run(): wrap the blocking loop in a self-healing
reconnect loop with backoff. It now returns ONLY when there are no local
subscribers (graceful teardown); any connection failure logs, backs off, re-reads
the desired channels, and re-subscribes.
Blast radius: the dedicated subscribe connection only (never the SSE pool, never
kiss). ssr Async + Server 136/136. Live (pl, 4 workers): held-open grid re-run
delivers fresh ui.grid.data frames on the open stream across multiple mutations.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ion's timeout Pure construction guard asserting RedisSubscribeConnectionFactory vends a connection with read_write_timeout: -1 (and password only when present) — so a regression that drops the idle-timeout fix fails here instead of silently in production after ~60s. A self-contained REAL blocking-loop + real-publish test was attempted but hangs: Swoole's Coroutine::cancel does not reliably interrupt Predis's blocking pubSubLoop read, so the loop cannot be stopped from a test without an injectable/closable connection or a stop signal on the subscriber. The real loop is proven LIVE; the testability refactor is the recommended follow-up. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Adds the {__ctrl:viewchange} control alongside {__ctrl:rerun}: handleControlFrame
branches on the control kind and shares the re-run+frame-write tail via
dispatchReRun(). A view-change resolves the latest view params (last-write-wins
from the new ViewChangeCoalescer; inline payload params as the no-coalescer
fallback), resolves the worker-local ReRunContext, and re-runs with a FILTER-ONLY
override forwarded to R2 — pushing the fresh frame on the SAME open fd.
submitViewChange() is the inbound intake the app's command endpoint calls: it
coalesces (latest-view-wins, collapse pending — the RerunCoalescer discipline,
kept on a SEPARATE counter so mutation and view-change re-runs never suppress
each other) and enqueues the control onto the session-addressed queue. The
ViewChangeCoalescer joins the pre-fork cross-worker shared-table plane
(TrackRSharedTables / CreateTrackRTablesListener) and is wired BEFORE the Redis
gate so the intra-instance view-change path works single-server.
Existing R4 test fakes updated to the new reRun(.. , $filterOverride) signature.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Stream Lifecycle Axis 2, Phase 1 — close the one residual SSE leak: a hard worker crash skips the per-connection finally, leaving the shared-mmap tier-1 SubscriptionTable row orphaned with no TTL. - SubscriptionTable: add a connected_at INT column (stamped at insert) and reapStaleConnections($maxAge, $now) that evicts rows older than cap+grace, returning their streaming_ids. Lone non-string column; the tier-separation invariant (no live subject) is unaffected. - ReapStaleSubscriptionsListener: a Timer::tick sweeper at WorkerStartFinalize (60s), modelled on DeferredRequestRegistry's GC. Threshold = SSE_MAX_CONNECTION_AGE_SECONDS + 60s grace, so a row past it cannot be live (the loop force-closes + reaps at the cap). Evicts the tier-1 row and clears the coupled RerunCoalescer pending mark. - Age-based ONLY, no pid/liveness probe: Swoole reuses worker_id on crash-restart, so a liveness check could false-positive on a row now owned by the restarted worker. - AsyncResourceSseServer: extract maxConnectionAgeSeconds() so the loop cap and the sweep threshold share one source of truth. Additive: onDisconnect/finally normal-drop reaping unchanged. No client or id-generation change. Unit suite over a real Swoole\Table proves orphan-reaped, live-untouched, and age-based; ai:verify pass (module_structure 0, phpstan_di 0). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…is 1b Phase 2) Stream Lifecycle Axis 1(b) Phase 2 — backend-only, back-compatible. The server now mints + owns the stream id at connect and announces it as a dedicated FIRST SSE event, one line before the initial data frame, while STILL accepting a client-sent stream_id as the addressing key (today's client is unchanged). - AsyncResourceSseServer::mintStreamId(): the single server-side source of new ids (CSPRNG, sse_<32hex> — the shape every keyed store already validates). - serveResourceStream() gains $serverStreamId: emits it as a ui.stream.id first frame before the initial data write; empty-$sessionId fallback upgraded from uniqid() to mintStreamId(). The data frame is untouched — the id rides its own event, so the byte-identical initial/re-run ui.grid.data invariant holds. - UiSseEventType::UiStreamId = 'ui.stream.id' so the typed chokepoint promotes it. Transition rule: a shape-valid client id still ADDRESSES the stream this phase; the distinct server-minted id is emitted for forward adoption (inert until the Phase-3 client reads it). When no client id is sent, announced == addressing key, so the two never diverge as live coordinates. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…e-liner
Live-on-events Phase 3a. ORM-backed grids go live for free (AggregateWriteEngine
auto-publishes ui.invalidate.{tenant}.{scopeKey}). A non-ORM / raw-DB write path
has no such chokepoint, so it had to hand-assemble the channel exactly like the
leads demo-add does. GridScopeInvalidator::touch($scope) collapses that to one
call, publishing the IDENTICAL channel/message the ORM auto-publish produces —
same ResourceInvalidationPublisher::channelFor() naming, same TenantContext
resolution, same ScopeInvalidationBusInterface transport — so the subscriber (R3)
re-runs identically whether the write came through the ORM or a raw query.
Container-managed (#[AsService] + #[InjectAsReadonly] bus); best-effort/data-less.
Proven in isolation by GridScopeInvalidatorTest (identical-channel + tenant scope
+ blank-scope no-op). Does not touch the subscribe/coalesce/re-run machine.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
… (One Way P4/P6)
- AbstractSseCollectionFeedHandler + SseCollectionFeedPayloadInterface: the
legacy grid-stream choreography re-homed on the canonical {data, meta}
envelope; ONE abstract seam buildCollectionResponse(); scopeKeys from core
#[WatchScopes]; SSE-path domain errors framed as ui.collection.error
- ui.collection.data / ui.collection.error in UiSseEventType; v1 ui.grid.data /
ui.grid.error members DELETED (now rejected strings, pinned by tests)
- LivePubSubChannelController: interrupt seam so a second distinct scope per
worker resubscribes the full set (loop snapshot was deaf to it);
ResourceInvalidationSubscriber::isSubscribedTo()/interrupt()
- SseKissPayload/AsyncResourceSseServer: server-minted stream id adoption +
rehydrate POST path for view changes on a held-open stream
- AbstractSseCollectionFeedHandler: AuthenticationException/AccessDeniedException from the builder now PROPAGATE instead of framing as ui.collection.error — initial connect 403s via the ExceptionMapper (a denied caller never mints a held-open stream + tier-1 row), a re-run tick TERMINATEs via reExecute() - ConnectCoordinator/ReapStaleSubscriptionsListener: also clear the ViewChangeCoalescer row on disconnect and in the orphan sweep — streaming ids are never reused, so a disconnect-before-drain row leaked for the table's life
|
Warning Review limit reached
More reviews will be available in 59 minutes and 14 seconds. Learn how PR review limits work. Your organization has run out of usage credits. Purchase more credits in the billing tab to continue. ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans include higher PR review limits than trial, open-source, and free plans. In all cases, reviews become available again over time. During sustained high-volume PR review activity, CodeRabbit may temporarily slow when the next review becomes available. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (56)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: e67a873a0c
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
Contributor
There was a problem hiding this comment.
Pull request overview
This PR advances the SSR/SSE “One Way / Track R” architecture by introducing a canonical SSE collection feed vocabulary, wiring the Track‑R consumer pipeline (subscription store + invalidation subscriber + rerun loop), and removing the legacy v1 grid/SSE endpoints and client script.
Changes:
- Add Track‑R subscription store primitives (tier‑1 shared table, tier‑2 worker registry, reverse index), publisher/subscriber invalidation pipeline, and lifecycle wiring (incl. reconnect + stale orphan reaper).
- Canonicalize SSE frame typing/event allow‑list (new
ui.stream.id,ui.collection.data,ui.collection.error) and introduce optional passthrough SSE event support. - Remove deprecated
/sseendpoint and legacy clientasync-resource-manager.jsasset wiring; add/expand unit tests for the new SSE/Track‑R behaviors.
Reviewed changes
Copilot reviewed 56 out of 56 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/Unit/UiEvent/UiSseEventTypeTest.php | Expands allow/deny lists to include new typed events and reject removed v1 ui.grid.* types. |
| tests/Unit/Server/Lifecycle/WireSseServedPathsListenerTest.php | Verifies SSE intercept is driven by transport: Sse routes, not hardcoded paths. |
| tests/Unit/Async/ViewChangeControlFrameTest.php | Exercises {__ctrl:viewchange} → rerun path and coalescing behavior. |
| tests/Unit/Async/ViewChangeCoalescerTest.php | Pins “latest view wins” semantics and consume/peek behaviors. |
| tests/Unit/Async/TypedSseFrameTest.php | Updates chokepoint reflection target (buildFrame) and adds passthrough/regression assertions. |
| tests/Unit/Async/SubscriptionStoreTest.php | Validates tier separation and reverse-index seam behavior over real Swoole tables. |
| tests/Unit/Async/StaleSubscriptionReaperTest.php | Covers age-based orphan reaping + coupled coalescer mark clearing. |
| tests/Unit/Async/ServerMintStreamIdTest.php | Pins server-minted stream id shape/entropy and allow-list membership. |
| tests/Unit/Async/ResourceInvalidationSubscriberTest.php | Proves routing, tenant isolation, coalescing, channel diffing, and dedicated subscribe connection invariants. |
| tests/Unit/Async/ResourceInvalidationPublisherTest.php | Proves tenant-scoped channel naming and sync execution pin enforcement. |
| tests/Unit/Async/RedisSubscribeConnectionFactoryTest.php | Guards read_write_timeout = -1 and password-setting behavior for dedicated subscriber connections. |
| tests/Unit/Async/HeldOpenGridStreamTest.php | Integration-style proof of publish→deliver→drain→rerun writing frames to the same held-open fd. |
| tests/Unit/Async/GridScopeInvalidatorTest.php | Verifies raw-DB invalidation helper publishes the same channel as ORM publisher. |
| tests/Unit/Async/ControlFrameReRunTest.php | Pins {__ctrl:rerun} behavior (rerun/terminate/noop/missing-context/etc.). |
| tests/Unit/Async/ConnectCoordinatorTest.php | Covers connect/disconnect tier population, coalescer clearing, and channel lifecycle behavior. |
| tests/Unit/Async/AsyncResourceSseServerTest.php | Adds coverage for generalized SSE dispatch based on served-path set. |
| tests/Unit/Application/Handler/PayloadHandler/AbstractSseCollectionFeedHandlerTest.php | Tests canonical envelope typing + watch-scope resolution + auth exception propagation. |
| src/Domain/Model/SubscriptionRecord.php | Introduces serialized cross-worker subscription record model. |
| src/Domain/Model/SubscriberRef.php | Introduces minimal routing coordinate returned by reverse index. |
| src/Domain/Contract/SubscriberIndexInterface.php | Defines reverse-index contract for resolving (tenant, scopeKey) to subscribers. |
| src/Domain/Contract/SseCollectionFeedPayloadInterface.php | Defines minimal payload contract for SSE collection feed handlers. |
| src/Domain/Contract/SessionControlDeliveryInterface.php | Defines contract for delivering control frames to session queues. |
| src/Domain/Contract/ScopeInvalidationBusInterface.php | Defines publisher-side invalidation bus contract (data-less publish). |
| src/Domain/Contract/ChannelSubscriptionControllerInterface.php | Defines connect-lifecycle seam to apply channel subscribe/unsubscribe deltas. |
| src/Application/Static/js/async-resource-manager.js | Deleted legacy async SSE manager client script. |
| src/Application/Static/assets.json | Removes deleted JS asset from asset registry. |
| src/Application/Service/UiEvent/UiSseEventType.php | Adds allow-listed typed events for stream id and canonical collection frames. |
| src/Application/Service/UiEvent/AsyncResourceSseMessagePublisher.php | Updates chokepoint reference and fences fan-out publish-to-user fail-closed. |
| src/Application/Service/Server/Lifecycle/WireTrackRConsumerListener.php | Wires Track‑R consumer-half (subscriber/coordinator/rerunner) per worker. |
| src/Application/Service/Server/Lifecycle/WireSseServedPathsListener.php | Collects transport: Sse route paths and registers them for SSE intercept dispatch. |
| src/Application/Service/Server/Lifecycle/TrackRSharedTables.php | Holds shared cross-worker tables (subscriptions + coalescers). |
| src/Application/Service/Server/Lifecycle/SsrBootstrapStateKey.php | Adds bootstrap key for Track‑R shared tables. |
| src/Application/Service/Server/Lifecycle/ReapStaleSubscriptionsListener.php | Adds periodic orphan subscription reaper (age-based) and coupled mark cleanup. |
| src/Application/Service/Server/Lifecycle/PinResourceInvalidationPublisherSyncListener.php | Adds boot-time guard ensuring publisher stays EventExecution::Sync. |
| src/Application/Service/Server/Lifecycle/CreateTrackRTablesListener.php | Creates Track‑R shared tables pre-fork. |
| src/Application/Service/Async/ViewChangeCoalescer.php | Implements cross-worker view-change coalescer with params slot. |
| src/Application/Service/Async/SwooleSseTransport.php | Adds Swoole implementation of the SseTransportInterface port. |
| src/Application/Service/Async/SubscriptionTable.php | Implements tier‑1 shared subscription store + stale-row reaping. |
| src/Application/Service/Async/SubscriptionDtoRegistry.php | Implements tier‑2 worker-local ReRunContext registry. |
| src/Application/Service/Async/SseSessionControlDelivery.php | Bridges control delivery interface to existing AsyncResourceSseServer::deliver(). |
| src/Application/Service/Async/SseRedisScopeInvalidationBus.php | Default bus binding to publish invalidations via existing Redis SSE mechanism. |
| src/Application/Service/Async/SsePassthroughEvent.php | Defines opt-in passthrough event key + allowed vocabulary for bare-body SSE protocols. |
| src/Application/Service/Async/ScanningSubscriberIndex.php | Implements reverse index via scan over subscription table. |
| src/Application/Service/Async/ResourceInvalidationSubscriber.php | Implements invalidation receiver (dedicated Redis subscribe loop + coalesced routing). |
| src/Application/Service/Async/ResourceInvalidationPublisher.php | Implements data-less invalidation publisher + sync execution pin. |
| src/Application/Service/Async/RerunCoalescer.php | Implements cross-worker “one pending rerun per stream” coalescer. |
| src/Application/Service/Async/RedisSubscribeConnectionFactory.php | Builds dedicated Predis clients with infinite read/write timeout for SUBSCRIBE. |
| src/Application/Service/Async/LivePubSubChannelController.php | Connect-lifecycle binding that ensures subscriber loop runs/resubscribes as needed. |
| src/Application/Service/Async/GridScopeInvalidator.php | Adds one-liner helper for raw DB writes to trigger invalidation publish. |
| src/Application/Service/Async/FanOutNotYetGatedException.php | Adds fail-closed barrier for fan-out SSE delivery methods until entitlement gating exists. |
| src/Application/Service/Async/ConnectCoordinator.php | Implements connect/disconnect lifecycle for Track‑R subscription tiers + channel reconciliation. |
| src/Application/Payload/Request/SseKissPayload.php | Declares KISS payload as transport: Sse and sets SSE gate model. |
| src/Application/Payload/Request/SseEndpointPayload.php | Deletes deprecated /sse payload. |
| src/Application/Handler/PayloadHandler/SseEndpointHandler.php | Deletes deprecated /sse handler. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Comment on lines
+83
to
+96
| // Intended Grid Model · Phase 2 — wire the view-change coalescer BEFORE the | ||
| // Redis gate. A view-change command is intra-instance (the browser POSTs to | ||
| // THIS deployment; the control rides the session-addressed queue, which falls | ||
| // back to the Swoole deliver-table when Redis is absent), so the view-change | ||
| // intake must work even single-server with no cross-instance bus. | ||
| AsyncResourceSseServer::setViewChangeCoalescer($tables->viewChangeCoalescer); | ||
|
|
||
| $connectionFactory = RedisSubscribeConnectionFactory::fromEnvironment(); | ||
| if ($connectionFactory === null) { | ||
| StaticLoggerBridge::debug('ssr', 'track_r_consumer_unwired', [ | ||
| 'reason' => 'no Redis configured — no cross-instance invalidation bus (view-change still wired)', | ||
| ]); | ||
| return; | ||
| } |
Member
Author
There was a problem hiding this comment.
Same root cause as the line-95 thread: the LOCAL collaborators (RouteReRunner + the $tables->coalescer rerun coalescer) need hoisting above the Redis gate so single-server / no-Redis mode wires setReRunner/setRerunCoalescer; only the subscriber + ConnectCoordinator are Redis-bound. Deferred for a validated follow-up.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
developtargetingmaster19commit(s) ahead,56changed file(s)Changes
Checks
bin/semitexa test:run <package>/tests, in-container, after review fixes): core 1042, api 95, orm 137, graphql 121, platform-ui 1087, ssr 354, UiPlayground 281, Playground 110 — all greenbin/semitexa ai:verifyon the sweep file set: 24 pass / 0 fail (syntax, module_structure, phpstan_di, targeted phpunit)ep-review-followups-20260612What to Review
src/Applicationsrc/Domaintests/UnitWhat to Verify