1package daemon
2
3import (
4 "context"
5 "sync"
6 "sync/atomic"
7
8 "github.com/bnema/vev/internal/ports"
9)
10
11type attachmentEffectPhase uint8
12
13const (
14 attachmentEffectsStable attachmentEffectPhase = iota
15 attachmentEffectsFrozen
16)
17
18// attachmentCapability is the single attachment-local publication consumed by
19// effect admission. It deliberately contains no independently mutable state:
20// a ticket is admitted only when every field matches the frame's token.
21type attachmentCapability struct {
22 sess *session
23 generation uint64
24 transport transportSnapshot
25 lease *attachmentLease
26}
27
28func capabilityFromToken(token attachmentConnectionToken) attachmentCapability {
29 return attachmentCapability{
30 sess: token.sess,
31 generation: token.generation,
32 transport: token.transport,
33 lease: token.lease,
34 }
35}
36
37func (c attachmentCapability) matches(token attachmentConnectionToken) bool {
38 return token.ac != nil && c.sess != nil && token.sess != nil && c.sess == token.sess &&
39 c.generation == token.generation && c.transport.transport == token.transport.transport &&
40 c.transport.incarnation == token.transport.incarnation && c.lease == token.lease
41}
42
43type attachmentEffectGate struct {
44 mu sync.Mutex
45 cond *sync.Cond
46 changed chan struct{}
47 phase attachmentEffectPhase
48 capability attachmentCapability
49 inFlight uint64
50 transportEffects map[*attachmentEffectTicket]transportSnapshot
51 failedTransport transportSnapshot
52 order atomic.Uint64
53}
54
55var nextAttachmentEffectGateOrder atomic.Uint64
56
57func (g *attachmentEffectGate) initLocked() {
58 if g.cond == nil {
59 g.cond = sync.NewCond(&g.mu)
60 }
61 if g.changed == nil {
62 g.changed = make(chan struct{})
63 }
64}
65
66func (g *attachmentEffectGate) signalLocked() {
67 g.initLocked()
68 close(g.changed)
69 g.changed = make(chan struct{})
70 g.cond.Broadcast()
71}
72
73func (g *attachmentEffectGate) immutableOrder() uint64 {
74 if order := g.order.Load(); order != 0 {
75 return order
76 }
77 order := nextAttachmentEffectGateOrder.Add(1)
78 if g.order.CompareAndSwap(0, order) {
79 return order
80 }
81 return g.order.Load()
82}
83
84func (g *attachmentEffectGate) inFlightCount() uint64 {
85 if g == nil {
86 return 0
87 }
88 g.mu.Lock()
89 defer g.mu.Unlock()
90 return g.inFlight
91}
92
93// attachmentEffectTicket reserves the exact published capability until End. End is
94// idempotent so error paths can safely defer it before doing any observable I/O.
95type attachmentEffectTicket struct {
96 gate *attachmentEffectGate
97 token attachmentConnectionToken
98 actionDaemon *Daemon
99 action string
100 ended atomic.Bool
101}
102
103func (t *attachmentEffectTicket) bindActionEnd(d *Daemon, action string) {
104 if t == nil || t.ended.Load() {
105 return
106 }
107 t.actionDaemon = d
108 t.action = action
109}
110
111func (t *attachmentEffectTicket) connectionToken() attachmentConnectionToken {
112 if t == nil {
113 return attachmentConnectionToken{}
114 }
115 token := t.token
116 token.effect = t
117 return token
118}
119
120func (t *attachmentEffectTicket) End() {
121 if t == nil || t.gate == nil || !t.ended.CompareAndSwap(false, true) {
122 return
123 }
124 g := t.gate
125 g.mu.Lock()
126 g.initLocked()
127 if g.inFlight == 0 {
128 g.mu.Unlock()
129 panic("daemon: attachment effect ticket ended without admission")
130 }
131 delete(g.transportEffects, t)
132 g.inFlight--
133 g.signalLocked()
134 g.mu.Unlock()
135 if t.actionDaemon != nil && t.actionDaemon.afterActionAttachmentEffectEnded != nil {
136 t.actionDaemon.afterActionAttachmentEffectEnded(t.action)
137 }
138}
139
140func (d *Daemon) endActionAttachmentEffect(effect *attachmentEffectTicket, _ string) {
141 if effect != nil {
142 effect.End()
143 }
144}
145
146// beginTransportSend marks only the interval spent in an external transport
147// send as interruptible. A replacement may close this exact published link to
148// release the send without retiring an idle healthy attachment.
149func (t *attachmentEffectTicket) beginTransportSend(expected transportSnapshot) bool {
150 if t == nil || t.gate == nil || t.ended.Load() || expected.transport == nil {
151 return false
152 }
153 g := t.gate
154 g.mu.Lock()
155 g.initLocked()
156 if t.ended.Load() || g.capability.transport.transport != expected.transport ||
157 g.capability.transport.incarnation != expected.incarnation {
158 g.mu.Unlock()
159 return false
160 }
161 if g.transportEffects == nil {
162 g.transportEffects = make(map[*attachmentEffectTicket]transportSnapshot)
163 }
164 g.transportEffects[t] = expected
165 g.signalLocked()
166 g.mu.Unlock()
167 return true
168}
169
170func (t *attachmentEffectTicket) reportTransportFailure(expected transportSnapshot) {
171 if t == nil || t.gate == nil || expected.transport == nil {
172 return
173 }
174 g := t.gate
175 g.mu.Lock()
176 if !t.ended.Load() && g.capability.transport.transport == expected.transport &&
177 g.capability.transport.incarnation == expected.incarnation {
178 g.failedTransport = expected
179 g.signalLocked()
180 }
181 g.mu.Unlock()
182}
183
184func (t *attachmentEffectTicket) endTransportSend() {
185 if t == nil || t.gate == nil {
186 return
187 }
188 g := t.gate
189 g.mu.Lock()
190 if _, exists := g.transportEffects[t]; exists {
191 delete(g.transportEffects, t)
192 g.signalLocked()
193 }
194 g.mu.Unlock()
195}
196
197// beginAttachmentEffect is the sole attachment-bound effect admission point. The gate is
198// held only long enough to reserve the capability; the effect itself must run
199// without the gate mutex.
200func (ac *attachedClient) beginAttachmentEffect(token attachmentConnectionToken) (*attachmentEffectTicket, bool) {
201 if ac == nil || token.ac != ac {
202 return nil, false
203 }
204 g := &ac.attachmentEffects
205 g.mu.Lock()
206 g.initLocked()
207 if g.phase != attachmentEffectsStable || !g.capability.matches(token) {
208 g.mu.Unlock()
209 return nil, false
210 }
211 g.inFlight++
212 g.mu.Unlock()
213 captured := token
214 captured.effect = nil
215 return &attachmentEffectTicket{gate: g, token: captured}, true
216}
217
218// beginCurrentAttachmentEffect waits out a transition that already froze ac, then
219// admits the capability derived from the session registry. Handshakes use it
220// after Welcome has completed so a replacement blocked behind that send can
221// publish before readiness or first paint.
222func (ac *attachedClient) beginCurrentAttachmentEffect(sess *session, tr ports.Transport) (attachmentConnectionToken, *attachmentEffectTicket, bool) {
223 //nolint:contextcheck // This compatibility wrapper deliberately delegates with a fresh context; cancellable callers use the context variant below.
224 return ac.beginCurrentAttachmentEffectContext(context.Background(), sess, tr)
225}
226
227func (ac *attachedClient) beginCurrentAttachmentEffectContext(ctx context.Context, sess *session, tr ports.Transport) (attachmentConnectionToken, *attachmentEffectTicket, bool) {
228 if ctx == nil {
229 ctx = context.Background()
230 }
231 if ac == nil || sess == nil || tr == nil {
232 return attachmentConnectionToken{}, nil, false
233 }
234 for {
235 token := sess.attachmentToken(ac, tr)
236 if token.ac == nil {
237 return token, nil, false
238 }
239 if ticket, admitted := ac.beginAttachmentEffect(token); admitted {
240 return token, ticket, true
241 }
242
243 g := &ac.attachmentEffects
244 g.mu.Lock()
245 g.initLocked()
246 if g.phase == attachmentEffectsFrozen {
247 changed := g.changed
248 g.mu.Unlock()
249 select {
250 case <-changed:
251 continue
252 case <-ctx.Done():
253 return token, nil, false
254 }
255 }
256 g.mu.Unlock()
257 // A stable mismatch is only recoverable when a publication raced token
258 // capture; an already-stale token proves that happened. A current token
259 // with a stable mismatch violates the publication invariant, so fail
260 // closed instead of spinning forever.
261 if token.current() {
262 return token, nil, false
263 }
264 // Publication may have completed between token capture and admission.
265 // Re-read the registry and exact generation instead of using stale authority.
266 }
267}
268
269// publishAttachmentCapability is used by direct/headless setup and by transition
270// publication while the gate is frozen. Production attachment changes publish only
271// through publishFrozenAttachmentCapability.
272func (ac *attachedClient) publishAttachmentCapability(token attachmentConnectionToken) {
273 if ac == nil {
274 return
275 }
276 g := &ac.attachmentEffects
277 g.mu.Lock()
278 g.initLocked()
279 g.capability = capabilityFromToken(token)
280 g.failedTransport = transportSnapshot{}
281 g.mu.Unlock()
282}
283
284// bootstrapAttachmentCapability supports direct/headless session construction. It
285// never changes an existing or frozen production publication.
286func (ac *attachedClient) bootstrapAttachmentCapability(token attachmentConnectionToken) {
287 if ac == nil || token.sess == nil || token.ac != ac {
288 return
289 }
290 g := &ac.attachmentEffects
291 g.mu.Lock()
292 g.initLocked()
293 if g.phase == attachmentEffectsStable && g.capability.sess == nil {
294 g.capability = capabilityFromToken(token)
295 g.failedTransport = transportSnapshot{}
296 }
297 g.mu.Unlock()
298}
299
300func (ac *attachedClient) publishFrozenAttachmentCapability(token attachmentConnectionToken) {
301 g := &ac.attachmentEffects
302 g.mu.Lock()
303 g.initLocked()
304 if g.phase != attachmentEffectsFrozen {
305 g.mu.Unlock()
306 panic("daemon: publishing attachment capability without transition freeze")
307 }
308 g.capability = capabilityFromToken(token)
309 g.failedTransport = transportSnapshot{}
310 g.mu.Unlock()
311}
312
313func (ac *attachedClient) invalidateFrozenAttachmentCapability() {
314 g := &ac.attachmentEffects
315 g.mu.Lock()
316 g.initLocked()
317 if g.phase != attachmentEffectsFrozen {
318 g.mu.Unlock()
319 panic("daemon: invalidating attachment capability without transition freeze")
320 }
321 g.capability = attachmentCapability{}
322 g.failedTransport = transportSnapshot{}
323 g.mu.Unlock()
324}