1package daemon
2
3import (
4 "context"
5 "sync"
6 "sync/atomic"
7
8 "github.com/bnema/vev/internal/ports"
9 "github.com/bnema/vev/internal/protocol"
10)
11
12type attachmentEffectPhase uint8
13
14const (
15 attachmentEffectsStable attachmentEffectPhase = iota
16 attachmentEffectsFrozen
17)
18
19type attachmentActivity uint8
20
21const (
22 attachmentActive attachmentActivity = iota
23 attachmentSuspended
24 attachmentActivating
25)
26
27// attachmentLifecycle owns the immutable committed capability and the
28// linearization gate for every attachment-bound observable effect. Transitions
29// freeze and drain it before changing membership, generation, transport,
30// activity, or render lease identity.
31type attachmentLifecycle struct {
32 activity attachmentActivity // protected by mu; independent of transition freeze ownership
33 retained protocol.ExactSessionTarget
34 generation atomic.Uint64
35 mu sync.Mutex
36 cond *sync.Cond
37 changed chan struct{}
38 phase attachmentEffectPhase
39 capability attachmentCapability
40 inFlight uint64
41 transportEffects map[*attachmentEffect]transportSnapshot
42 failedTransport transportSnapshot
43 order atomic.Uint64
44}
45
46var nextAttachmentEffectGateOrder atomic.Uint64
47
48func (ac *attachedClient) attachmentActivity() attachmentActivity {
49 ac.lifecycle.mu.Lock()
50 defer ac.lifecycle.mu.Unlock()
51 return ac.lifecycle.activity
52}
53
54func (g *attachmentLifecycle) generationValue() uint64 {
55 if g == nil {
56 return 0
57 }
58 return g.generation.Load()
59}
60
61func (g *attachmentLifecycle) initLocked() {
62 if g.cond == nil {
63 g.cond = sync.NewCond(&g.mu)
64 }
65 if g.changed == nil {
66 g.changed = make(chan struct{})
67 }
68}
69
70func (g *attachmentLifecycle) signalLocked() {
71 g.initLocked()
72 close(g.changed)
73 g.changed = make(chan struct{})
74 g.cond.Broadcast()
75}
76
77func (g *attachmentLifecycle) immutableOrder() uint64 {
78 if order := g.order.Load(); order != 0 {
79 return order
80 }
81 order := nextAttachmentEffectGateOrder.Add(1)
82 if g.order.CompareAndSwap(0, order) {
83 return order
84 }
85 return g.order.Load()
86}
87
88func (g *attachmentLifecycle) inFlightCount() uint64 {
89 if g == nil {
90 return 0
91 }
92 g.mu.Lock()
93 defer g.mu.Unlock()
94 return g.inFlight
95}
96
97// attachmentEffect is an explicit admission of one immutable capability. It
98// remains valid until End because transitions freeze admission and drain all
99// existing effects before changing lifecycle identity. End is idempotent.
100type attachmentEffect struct {
101 attachmentCapability
102 lifecycle *attachmentLifecycle
103 actionDaemon *Daemon
104 action string
105 // uiActionID belongs to this admitted input, not the attachment lifecycle.
106 // Set before routing and copy only into its exact delayed continuation.
107 uiActionID uint64
108 ended atomic.Bool
109}
110
111func (t *attachmentEffect) bindActionEnd(d *Daemon, action string) {
112 if t == nil || t.ended.Load() {
113 return
114 }
115 t.actionDaemon = d
116 t.action = action
117}
118
119func (t *attachmentEffect) capability() attachmentCapability {
120 if t == nil {
121 return attachmentCapability{}
122 }
123 return t.attachmentCapability
124}
125
126func (t *attachmentEffect) current() bool {
127 return t != nil && !t.ended.Load()
128}
129
130func (t *attachmentEffect) currentSessionLocked() bool {
131 return t.current()
132}
133
134func (t *attachmentEffect) End() {
135 if t == nil || t.lifecycle == nil || !t.ended.CompareAndSwap(false, true) {
136 return
137 }
138 g := t.lifecycle
139 g.mu.Lock()
140 g.initLocked()
141 if g.inFlight == 0 {
142 g.mu.Unlock()
143 panic("daemon: attachment effect ended without admission")
144 }
145 delete(g.transportEffects, t)
146 g.inFlight--
147 g.signalLocked()
148 g.mu.Unlock()
149 if t.actionDaemon != nil && t.actionDaemon.afterActionAttachmentEffectEnded != nil {
150 t.actionDaemon.afterActionAttachmentEffectEnded(t.action)
151 }
152}
153
154func (d *Daemon) endActionAttachmentEffect(effect *attachmentEffect, _ string) {
155 if effect != nil {
156 effect.End()
157 }
158}
159
160// beginTransportSend marks only the interval spent in an external transport
161// send as interruptible. A replacement may close this exact published link to
162// release the send without retiring an idle healthy attachment.
163func (t *attachmentEffect) beginTransportSend(expected transportSnapshot) bool {
164 if t == nil || t.lifecycle == nil || t.ended.Load() || expected.transport == nil {
165 return false
166 }
167 g := t.lifecycle
168 g.mu.Lock()
169 g.initLocked()
170 if t.ended.Load() || g.capability.transport.transport != expected.transport ||
171 g.capability.transport.incarnation != expected.incarnation {
172 g.mu.Unlock()
173 return false
174 }
175 if g.transportEffects == nil {
176 g.transportEffects = make(map[*attachmentEffect]transportSnapshot)
177 }
178 g.transportEffects[t] = expected
179 g.signalLocked()
180 g.mu.Unlock()
181 return true
182}
183
184func (t *attachmentEffect) reportTransportFailure(expected transportSnapshot) {
185 if t == nil || t.lifecycle == nil || expected.transport == nil {
186 return
187 }
188 g := t.lifecycle
189 g.mu.Lock()
190 if !t.ended.Load() && g.capability.transport.transport == expected.transport &&
191 g.capability.transport.incarnation == expected.incarnation {
192 g.failedTransport = expected
193 g.signalLocked()
194 }
195 g.mu.Unlock()
196}
197
198func (t *attachmentEffect) endTransportSend() {
199 if t == nil || t.lifecycle == nil {
200 return
201 }
202 g := t.lifecycle
203 g.mu.Lock()
204 if _, exists := g.transportEffects[t]; exists {
205 delete(g.transportEffects, t)
206 g.signalLocked()
207 }
208 g.mu.Unlock()
209}
210
211func (t *attachmentEffect) sendControl(message protocol.ServerMessage) error {
212 if !t.current() || t.ac == nil || t.transport.transport == nil {
213 return errAttachmentTransition
214 }
215 t.ac.sendMu.Lock()
216 defer t.ac.sendMu.Unlock()
217 if !t.current() || !t.ac.transportSnapshotCurrent(t.transport) || !t.beginTransportSend(t.transport) {
218 return errAttachmentTransition
219 }
220 // Bind navigation at its single admitted send boundary. Palette catalogue
221 // values remain reusable; a later action must never inherit an earlier
222 // template's cause or consult a mutable attachment-wide "latest action".
223 switch navigation := message.(type) {
224 case protocol.AttachTarget:
225 navigation.CauseActionID = t.uiActionID
226 message = navigation
227 case protocol.RouteNavigationAction:
228 navigation.CauseActionID = t.uiActionID
229 message = navigation
230 case protocol.RouteCreateSessionAction:
231 navigation.CauseActionID = t.uiActionID
232 message = navigation
233 case protocol.NavigationInventorySelection:
234 navigation.CauseActionID = t.uiActionID
235 message = navigation
236 }
237 err := t.transport.transport.SendServer(message)
238 if err != nil {
239 t.reportTransportFailure(t.transport)
240 }
241 t.endTransportSend()
242 return err
243}
244
245// beginAttachmentEffect is the sole attachment-bound effect admission point. The gate is
246// held only long enough to reserve the capability; the effect itself must run
247// without the gate mutex.
248func (ac *attachedClient) beginAttachmentEffect(token attachmentCapability) (*attachmentEffect, bool) {
249 if ac == nil || token.ac != ac || !token.current() {
250 return nil, false
251 }
252 g := &ac.lifecycle
253 g.mu.Lock()
254 g.initLocked()
255 if g.phase != attachmentEffectsStable || g.activity != attachmentActive || !g.capability.sameIdentity(token) {
256 g.mu.Unlock()
257 return nil, false
258 }
259 g.inFlight++
260 g.mu.Unlock()
261 return &attachmentEffect{attachmentCapability: token, lifecycle: g}, true
262}
263
264// beginCurrentAttachmentEffect waits out a transition that already froze ac, then
265// admits the capability derived from the session registry. Handshakes use it
266// after Welcome has completed so a replacement blocked behind that send can
267// publish before readiness or first paint.
268func (ac *attachedClient) beginCurrentAttachmentEffect(sess *session, tr ports.ServerConnection) (attachmentCapability, *attachmentEffect, bool) {
269 //nolint:contextcheck // This compatibility wrapper deliberately delegates with a fresh context; cancellable callers use the context variant below.
270 return ac.beginCurrentAttachmentEffectContext(context.Background(), sess, tr)
271}
272
273func (ac *attachedClient) beginCurrentAttachmentEffectContext(ctx context.Context, sess *session, tr ports.ServerConnection) (attachmentCapability, *attachmentEffect, bool) {
274 if ctx == nil || ac == nil || sess == nil || tr == nil {
275 return attachmentCapability{}, nil, false
276 }
277 for {
278 token := sess.captureAttachmentCapability(ac, tr)
279 if token.ac == nil {
280 return token, nil, false
281 }
282 if ticket, admitted := ac.beginAttachmentEffect(token); admitted {
283 return token, ticket, true
284 }
285
286 g := &ac.lifecycle
287 g.mu.Lock()
288 g.initLocked()
289 if g.phase == attachmentEffectsFrozen {
290 changed := g.changed
291 g.mu.Unlock()
292 select {
293 case <-changed:
294 continue
295 case <-ctx.Done():
296 return token, nil, false
297 }
298 }
299 g.mu.Unlock()
300 // A stable mismatch is only recoverable when a publication raced capability
301 // capture; an already-stale capability proves that happened. A current capability
302 // with a stable mismatch violates the publication invariant, so fail
303 // closed instead of spinning forever.
304 if token.current() {
305 return token, nil, false
306 }
307 // Publication may have completed between token capture and admission.
308 // Re-read the registry and exact generation instead of using stale authority.
309 }
310}
311
312// installInitialCapability publishes only an empty lifecycle owner. It supports
313// direct/headless construction without creating a second mutable publication.
314func (g *attachmentLifecycle) installInitialCapability(capability attachmentCapability) {
315 g.mu.Lock()
316 g.initLocked()
317 if g.phase == attachmentEffectsStable && g.capability.sess == nil {
318 g.capability = capability
319 g.failedTransport = transportSnapshot{}
320 }
321 g.mu.Unlock()
322}
323
324func (ac *attachedClient) publishFrozenAttachmentCapability(capability attachmentCapability) attachmentCapability {
325 g := &ac.lifecycle
326 g.mu.Lock()
327 g.initLocked()
328 if g.phase != attachmentEffectsFrozen {
329 g.mu.Unlock()
330 panic("daemon: publishing attachment capability without transition freeze")
331 }
332 capability.generation = g.generation.Add(1)
333 g.capability = capability
334 g.failedTransport = transportSnapshot{}
335 g.mu.Unlock()
336 return capability
337}
338
339// invalidateRetiredAttachmentCapability retires parked identity either while
340// stable with no admitted effects or under a terminal teardown's existing
341// freeze. Live membership transitions use invalidateFrozenAttachmentCapability.
342func (ac *attachedClient) invalidateRetiredAttachmentCapability() {
343 g := &ac.lifecycle
344 g.mu.Lock()
345 g.initLocked()
346 if g.phase != attachmentEffectsFrozen && (g.phase != attachmentEffectsStable || g.inFlight != 0) {
347 g.mu.Unlock()
348 panic("daemon: invalidating an active attachment capability")
349 }
350 g.generation.Add(1)
351 g.capability = attachmentCapability{}
352 g.failedTransport = transportSnapshot{}
353 g.mu.Unlock()
354}
355
356func (ac *attachedClient) invalidateFrozenAttachmentCapability() {
357 g := &ac.lifecycle
358 g.mu.Lock()
359 g.initLocked()
360 if g.phase != attachmentEffectsFrozen {
361 g.mu.Unlock()
362 panic("daemon: invalidating attachment capability without transition freeze")
363 }
364 g.generation.Add(1)
365 g.capability = attachmentCapability{}
366 g.failedTransport = transportSnapshot{}
367 g.mu.Unlock()
368}
369
370// detachFrozenAttachmentLocked atomically unregisters ac from sess and
371// invalidates its frozen capability. Callers must already hold ac's transition
372// freeze and sess.mu (plus routing locks where routing discipline applies); it
373// never freezes, drains, waits, or parks, so terminal teardown and the
374// daemon-owned suspension expiry can share one exact detach commit. A false
375// result means another path already detached ac. Only the winner performs
376// geometry reconciliation and transport/resource cleanup, outside locks.
377func detachFrozenAttachmentLocked(sess *session, ac *attachedClient) bool {
378 if sess == nil || ac == nil || ac.currentAttachmentSession() != sess || !attachmentRegisteredLocked(sess, ac) {
379 return false
380 }
381 sess.unregisterAttachmentLocked(ac)
382 ac.setSession(nil)
383 ac.invalidateFrozenAttachmentCapability()
384 return true
385}