1package brokeripc
2
3import (
4 "context"
5 "errors"
6
7 "github.com/bnema/vev/internal/adapters/brokerwire"
8 "github.com/bnema/vev/internal/domain"
9 "github.com/bnema/vev/internal/ports"
10)
11
12// Snapshot publication and mutating operations.
13//
14// A Subscribe frame starts one publisher goroutine for exactly that connection
15// and generation. The publisher waits on the admitted core service's coalescing
16// subscription, reads the newest publication, and writes it as one multipart
17// transfer. A slow client therefore blocks only its own publisher: the core
18// subscription coalesces, so the next wake reads the newest snapshot rather than
19// replaying a backlog, and no other connection is affected. Each generation owns
20// its wake channel, so a superseded publisher can never consume the wake that
21// was meant for its successor, and a resubscribe or resync always publishes the
22// current snapshot for the generation that asked.
23//
24// Mutating operations run one goroutine each, bounded by the brokerwire
25// per-connection pending-operation tracker, and complete exactly once with a
26// typed result. The request context is the connection's, so shutdown cancels an
27// operation that is still in flight.
28
29// publisher is one subscription generation's publication series: the core
30// subscription it drains, the context that ends it, and the capacity-one wake
31// channel that coalesces demands for it. The wake is owned by exactly one
32// generation, so a superseded publisher can never drain a later generation's
33// demand.
34type publisher struct {
35 generation brokerwire.SubscriptionGeneration
36 sub ports.BrokerSubscription
37 cancel context.CancelFunc
38 wake chan struct{}
39}
40
41// passiveSubscriber is the optional core capability behind a passive
42// Subscribe: a read-only snapshot subscription that never counts as
43// observation demand. A core without it serves every subscription as demand.
44type passiveSubscriber interface {
45 SubscribePassive() (ports.BrokerSubscription, error)
46}
47
48// retargetPublisher replaces any active publisher with one for the new
49// generation and publishes the current snapshot immediately. A failed
50// subscription ends the superseded series first, so no publisher survives to
51// publish under a generation the connection tracker already retired.
52func (s *serverSession) retargetPublisher(generation brokerwire.SubscriptionGeneration, passive bool) error {
53 subscribe := s.core.Subscribe
54 if reader, ok := s.core.(passiveSubscriber); passive && ok {
55 subscribe = reader.SubscribePassive
56 }
57 sub, err := subscribe()
58 if err != nil {
59 s.stopPublisher()
60 detail := errorDetail(err)
61 return s.send(brokerwire.BrokerErrorMessage{Epoch: s.epoch, Connection: s.scope.Connection, Error: detail})
62 }
63 s.stopPublisher()
64 ctx, cancel := context.WithCancel(s.ctx)
65 pub := &publisher{generation: generation, sub: sub, cancel: cancel, wake: make(chan struct{}, 1)}
66 s.subMu.Lock()
67 s.pub = pub
68 s.subMu.Unlock()
69 s.wg.Add(1)
70 go func() {
71 defer s.wg.Done()
72 defer sub.Close()
73 s.publishLoop(ctx, pub)
74 }()
75 s.wakePublisher()
76 return nil
77}
78
79// publishLoop publishes the newest committed snapshot on every coalesced wake
80// for one generation until it is superseded or the session ends.
81func (s *serverSession) publishLoop(ctx context.Context, pub *publisher) {
82 for {
83 select {
84 case <-ctx.Done():
85 return
86 case <-pub.wake:
87 case <-pub.sub.Changed():
88 }
89 if err := s.publishSnapshot(pub.generation); err != nil {
90 // A publication failure is a carriage failure: settle the
91 // connection so its reader observes the terminal cause.
92 s.abort(err)
93 return
94 }
95 }
96}
97
98// publishSnapshot writes one multipart transfer for the active generation.
99func (s *serverSession) publishSnapshot(generation brokerwire.SubscriptionGeneration) error {
100 snapshot := s.core.Snapshot()
101 if snapshot.Epoch == 0 {
102 // The core has published nothing yet; there is no transfer to send.
103 return nil
104 }
105 if snapshot.Epoch != s.epoch {
106 // A publication from another broker incarnation can never be applied
107 // to this connection's scope.
108 return nil
109 }
110 parts, err := snapshotParts(snapshot, s.epoch, s.scope.Connection, generation)
111 if err != nil {
112 return err
113 }
114 total, err := snapshotTransferBytes(parts, s.ceilings.MaxReceiveEnvelopeBytes, s.ceilings.StreamChunkLimit)
115 if err != nil {
116 return err
117 }
118 if total > uint64(brokerwire.MaxSnapshotStagedBytes) {
119 return errors.New("brokeripc: snapshot publication exceeds the staged ceiling")
120 }
121 for _, part := range parts {
122 if err := s.send(part); err != nil {
123 return err
124 }
125 }
126 return nil
127}
128
129// stopPublisher ends the active publication series, if any. The superseded
130// publisher observes the canceled context and its closed subscription, and only
131// its own wake channel could ever wake it, so it can never drain a demand meant
132// for a later generation.
133func (s *serverSession) stopPublisher() {
134 s.subMu.Lock()
135 pub := s.pub
136 s.pub = nil
137 s.subMu.Unlock()
138 if pub == nil {
139 return
140 }
141 pub.cancel()
142 pub.sub.Close()
143}
144
145// wakePublisher coalesces one publication demand on the active generation's own
146// wake channel. A demand raised while no publisher is active is dropped: the
147// next subscription publishes the current snapshot as its first action.
148func (s *serverSession) wakePublisher() {
149 s.subMu.Lock()
150 pub := s.pub
151 s.subMu.Unlock()
152 if pub == nil {
153 return
154 }
155 select {
156 case pub.wake <- struct{}{}:
157 default:
158 }
159}
160
161// startMutation admits one mutating operation under the brokerwire pending
162// bound and runs it in its own goroutine. A duplicate admission is dropped:
163// exactly one completion ever travels for one operation identity, and the peer's
164// own tracker already fences replays.
165func (s *serverSession) startMutation(operation ports.BrokerOperationID, kind brokerwire.RegisterMutationKind, run func(context.Context) (domain.RemoteRegistration, bool, error)) error {
166 if err := s.conn.AdmitOperation(operation); err != nil {
167 switch {
168 case errors.Is(err, brokerwire.ErrOperationCompleted), errors.Is(err, brokerwire.ErrOperationPending):
169 return nil
170 case errors.Is(err, brokerwire.ErrTooManyPendingOperations):
171 detail := errorDetail(ports.BrokerAdmissionLimit)
172 return s.send(brokerwire.BrokerErrorMessage{Epoch: s.epoch, Connection: s.scope.Connection, Error: detail})
173 default:
174 return errors.Join(ErrProtocol, err)
175 }
176 }
177 s.wg.Add(1)
178 go func() {
179 defer s.wg.Done()
180 registration, removed, err := run(s.ctx)
181 outcome := ports.BrokerOutcomeOK
182 if err != nil {
183 outcome = ports.BrokerOutcomeFailed
184 var unknown ports.BrokerStoreOutcomeUnknownError
185 var unknownPointer *ports.BrokerStoreOutcomeUnknownError
186 if errors.As(err, &unknown) || errors.As(err, &unknownPointer) {
187 outcome = ports.BrokerOutcomeUnknown
188 }
189 registration = domain.RemoteRegistration{}
190 removed = false
191 }
192 _ = s.conn.CompleteOperation(operation, outcome)
193 if s.ctx.Err() != nil {
194 return
195 }
196 detail := errorDetail(err)
197 result := brokerwire.OperationResult{
198 Epoch: s.epoch, Connection: s.scope.Connection, Operation: operation,
199 Outcome: outcome, Removed: removed, Registration: registration,
200 Error: detail, HasError: detail != (brokerwire.ErrorDetail{}),
201 }
202 if err := result.ValidateForMutation(kind); err != nil {
203 s.abort(errors.Join(ErrProtocol, err))
204 return
205 }
206 if err := s.send(result); err != nil {
207 s.abort(err)
208 }
209 }()
210 return nil
211}