refactor: extract world-free runtime services - #941
Conversation
|
Claude finished @winton-library's task in 5m 37s —— View job Code Review
I read the extracted code against the pre-refactor Four findings, all posted inline. Three of them are pre-existing behavior carried over verbatim — I'm raising them because this PR is what promotes
Smaller things, not worth inline comments:
Verification
|
capy-ai
Bot
changed the title
![capy-ai[bot]](https://avatars.githubusercontent.com/in/1915919?s=60&v=4){kind=small}
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| // PublishEvent delivers evt to everyone waiting for it: any SendCommandWithReply call awaiting an | ||
| // event of this name, and every stream subscriber whose subscriptions match it. | ||
| // | ||
| // A subscriber whose stream is closed or failing is skipped and logged rather than failing the | ||
| // call, since one broken client must not stop an event reaching the rest. | ||
| // | ||
| //nolint:gocognit // Put everything here so you can understand the logic in one place. | ||
| func (s *Server) PublishEvent(evt Event) error { | ||
| eventPb := &iscv1.Event{ | ||
| Name: evt.Name, | ||
| Payload: evt.Payload, | ||
| } | ||
|
|
||
| s.mu.RLock() | ||
| var subscribers []*streamSubscriber | ||
| //nolint:nestif // It's fine | ||
| if evt.Recipient != "" { | ||
| if subscriber, exists := s.subscribers[evt.Recipient]; exists { | ||
| for subscription := range subscriber.events { | ||
| if matchesEvent(subscription, eventPb.GetName()) { | ||
| subscribers = []*streamSubscriber{subscriber} | ||
| break | ||
| } | ||
| } | ||
| } else { | ||
| s.log.Debug().Str("recipient", evt.Recipient).Str("event", eventPb.GetName()).Msg("recipient has no open stream") | ||
| } | ||
| } else { | ||
| subscribers = make([]*streamSubscriber, 0, len(s.subscribers)) | ||
| for _, subscriber := range s.subscribers { | ||
| for subscription := range subscriber.events { | ||
| if matchesEvent(subscription, eventPb.GetName()) { | ||
| subscribers = append(subscribers, subscriber) | ||
| break | ||
| } | ||
| } | ||
| } | ||
| } | ||
| waiters := append([]chan *iscv1.Event(nil), s.replyWaiters[eventPb.GetName()]...) |
There was a problem hiding this comment.
Reply waiters ignore Recipient — a directed event reaches other users' in-flight replies
subscribers is filtered by evt.Recipient (lines 425–446), but waiters is not: every in-flight SendCommandWithReply registered under this event name gets the payload, no matter who the event was addressed to.
Failure scenario: user B calls SendCommandWithReply{EventName: "loot_awarded"} and blocks. Concurrently the runtime publishes Event{Name: "loot_awarded", Recipient: "A"} for user A. B's waiter receives A's payload and returns it as B's reply. Event names are part of a game's public schema, so B doesn't need to guess anything.
This is carried over verbatim from pkg/cardinal/service.go on main, so it isn't a regression — but this PR is what makes pkg/shard the single normative definition of the protocol and adds a conformance suite for it, which makes it the right moment to either fix it or pin it deliberately. A recipient field on the waiter, compared against evt.Recipient (empty = broadcast, matches all), would close it.
| if err := s.commands.Submit(ctx, cmd); err != nil { | ||
| return connect.NewError(connect.CodeInvalidArgument, eris.Wrap(err, "failed to enqueue command")) | ||
| } |
There was a problem hiding this comment.
The two entry points still disagree on how a sink failure is reported
This PR deliberately moved the ISC path to codes.Internal (pkg/micro/isc.go:92) on the reasoning that a runtime refusing a command is not a malformed request. The client path keeps CodeInvalidArgument for the exact same Submit error.
Under Cardinal today this is unreachable in practice (Enqueue only errors on an unregistered name), so nothing regresses. But CommandSink's own doc — "running it now inside a transaction" — is the case this refactor exists to enable, and there Submit fails for storage errors, aborted transactions, backpressure. All of those would be reported to the client as InvalidArgument: a permanent, don't-retry code for a transient server-side fault, and a different code than the same failure gets over ISC.
A seam that lets the sink pick works nicely, since connect.CodeOf returns CodeUnknown for anything that isn't already a *connect.Error:
if err := s.commands.Submit(ctx, cmd); err != nil {
if connect.CodeOf(err) != connect.CodeUnknown {
return err // the sink chose its own code
}
return connect.NewError(connect.CodeInternal, eris.Wrap(err, "failed to accept command"))
}Whichever way you go, TestSendCommand_SinkErrorRejectsRequest (server_internal_test.go:128) currently pins InvalidArgument for a sink that returns "runtime refused", so the conformance suite is asserting the behavior in question.
| func (s *Server) subscribeEvents(user *auth.User, subscriptions []*cardinalv1.EventSubscription) { | ||
| s.mu.Lock() | ||
| defer s.mu.Unlock() | ||
|
|
||
| subscriber := s.subscribers[user.ID] | ||
| assert.That(subscriber != nil, "subscriber should exist for authenticated stream") | ||
|
|
||
| for _, subscription := range subscriptions { | ||
| for _, eventName := range subscription.GetEvents() { | ||
| subscriber.events[eventName] = struct{}{} | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (s *Server) unsubscribeEvents(user *auth.User, subscriptions []*cardinalv1.EventSubscription) { | ||
| s.mu.Lock() | ||
| defer s.mu.Unlock() | ||
|
|
||
| subscriber := s.subscribers[user.ID] | ||
| assert.That(subscriber != nil, "subscriber should exist for authenticated stream") | ||
|
|
||
| for _, subscription := range subscriptions { | ||
| for _, eventName := range subscription.GetEvents() { | ||
| delete(subscriber.events, eventName) | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
TOCTOU: the hasSubscriber check and the map mutation don't share a lock
SubscribeEvents/UnsubscribeEvents take RLock for hasSubscriber (line 300 / 319), release it, then take Lock here. If the client's StartEventStream returns in that window, defer s.removeSubscriber(user) has already deleted the entry and subscriber is nil.
In a normal build assert.That panics (pkg/assert panics on !cond); with -tags release it's a no-op and line 378 nil-derefs instead. Either way it's a panic in a request handler triggered by an ordinary client disconnect racing its own subscribe — net/http recovers it, but the connection dies and the assertion's "should exist" invariant simply isn't one.
Carried over from pkg/cardinal/service.go, so not a regression. Cheap to make actually true, though: have subscribeEvents/unsubscribeEvents do the lookup under their own write lock and return false when the subscriber is gone, and drop the separate hasSubscriber pre-check — that also removes the double locking.
| t.Run("marshals and publishes", func(t *testing.T) { | ||
| t.Parallel() | ||
| prng := testutils.NewRand(t) | ||
| fixture := newServiceFixture(t, prng, false) | ||
|
|
||
| payload := testutils.SimpleEvent{Value: prng.Int()} | ||
| waiter := fixture.svc.addReplyWaiter(payload.Name()) | ||
| defer fixture.svc.removeReplyWaiter(payload.Name(), waiter) | ||
| require.NoError(t, fixture.svc.publishDefaultEvent(event.Event{ | ||
| Kind: event.KindDefault, | ||
| Payload: testutils.SimpleEvent{Value: prng.Int()}, | ||
| })) | ||
| }) |
There was a problem hiding this comment.
Payload round-trip integrity is no longer asserted anywhere
The test this replaces registered a waiter, published, and then decoded the delivered bytes back through SimpleEvent{}.UnmarshalWire and compared them to the original. This version only asserts publishDefaultEvent returns nil, so it would pass if MarshalWire emitted bytes no client could decode.
The pkg/shard suite doesn't cover the gap either — it proves opaque []byte travels from PublishEvent to a waiter, which is the right split, but nobody now checks that publishDefaultEvent's marshaling is what a client can read. Keeping the decode-and-compare here (against s.shardServer.addReplyWaiter, or by asserting on the shard.Event handed to a stub EventBus) restores it without pulling protocol assertions back into pkg/cardinal.
1 participant
Decompose Cardinal so a world-free runtime can serve the same protocol
Everything a client can observe about the CardinalService protocol lived inside
pkg/cardinal, welded to a world. ADR-057 needs a tickless shard to serve that same protocol, which would have meant reimplementing it — and every disagreement between the two copies would be a client-visible bug.Four extractions, no behavior change for existing worlds:
pkg/plugin/data—Plugin.Load(). Loading config never needed a world, only snapshot reconciliation did.pkg/auth—User,Mode,NewMiddleware, both authenticators.pkg/micro—Service.ServeCommands/Client.SendCommand. Inbound ISC was ~30 lines of validation ending in one line of runtime work.pkg/shard— the CardinalService handler, behindCommandSink+EventBus.pkg/cardinal/service.go: 759 → 248 lines. Cardinal's implementation of the seam is fourlines, and the same method now serves the ISC path, so client and inter-shard commands
converge on one implementation instead of two that happen to agree.
Fixes a latent reply-waiter race
SendCommandWithReplyregistered its waiter after submitting the command. Harmless under a tick — the command waits in a queue, so the waiter always wins. A runtime that executes insideSubmitpublishes the reply before returning, so the event lands with nobody listening and the client blocks until timeout. The shared server registers first; regression test confirmed by restoring the old order (fails withcontext deadline exceeded).Behavior changes
Internalrather thanInvalidArgument. Only fires for an unregistered command name, which is unreachable through a registered endpoint.Compatibility
cardinal.User,AuthMode,ParseAuthMode,UserFromContextremain as aliases — existing game code compiles unchanged.newServicenow returns an error (3 internal call sites updated).Stack created with GitHub Stacks CLI • Give Feedback 💬