1package daemon
2
3import (
4 "sort"
5 "sync"
6 "sync/atomic"
7
8 "github.com/bnema/vev/internal/ports"
9)
10
11type attachmentEffectDrainDeadline struct {
12 clock ports.Clock
13 done chan struct{}
14 stopCh chan struct{}
15 finished chan struct{}
16 startOnce sync.Once
17 stopOnce sync.Once
18 started atomic.Bool
19}
20
21func newAttachmentEffectDrainDeadline(clock ports.Clock) *attachmentEffectDrainDeadline {
22 return &attachmentEffectDrainDeadline{
23 clock: clock, done: make(chan struct{}), stopCh: make(chan struct{}), finished: make(chan struct{}),
24 }
25}
26
27func (d *attachmentEffectDrainDeadline) stop() {
28 if d == nil {
29 return
30 }
31 d.stopOnce.Do(func() {
32 // Claim an unstarted deadline without allocating a timer. Done is never
33 // called after the owning freeze/drain operation returns.
34 d.startOnce.Do(func() {})
35 if !d.started.Load() {
36 return
37 }
38 close(d.stopCh)
39 <-d.finished
40 })
41}
42
43func (d *attachmentEffectDrainDeadline) Done() <-chan struct{} {
44 if d == nil {
45 return nil
46 }
47 d.startOnce.Do(func() {
48 d.started.Store(true)
49 timer := d.clock.NewTimer(detachNotifyTimeout)
50 go func() {
51 defer close(d.finished)
52 select {
53 case <-timer.C():
54 close(d.done)
55 case <-d.stopCh:
56 timer.Stop()
57 }
58 }()
59 })
60 return d.done
61}
62
63type attachmentTransitionGuard struct {
64 clients []*attachedClient
65 interruptions map[*attachedClient]transportSnapshot
66 acquired bool
67 drained bool
68}
69
70type attachmentTransportInterrupt struct {
71 ac *attachedClient
72 transport transportSnapshot
73}
74
75// attachmentEffectFreezeOptions controls one attachment-effect-gate freeze operation. A freeze
76// must be called without daemon, routing, session, coordinator, send, overlay,
77// router, or pane locks held.
78type attachmentEffectFreezeOptions struct {
79 interrupts []attachmentTransportInterrupt
80 done func() <-chan struct{}
81 afterFrozen func(*attachedClient)
82 nonblocking bool
83}
84
85// freezeAttachmentEffectGates establishes transition priority in immutable attachment
86// order, then drains earlier admissions.
87func freezeAttachmentEffectGates(clients ...*attachedClient) attachmentTransitionGuard {
88 return freezeAttachmentEffectGatesWith(attachmentEffectFreezeOptions{}, clients...)
89}
90
91// freezeAttachmentEffectGatesWith uses done as one overall bound for ordered
92// acquisition and drain. Acquisition is all-or-nothing: if the bound expires
93// behind another owner, only gates acquired by this call are rolled back, in
94// reverse order. nonblocking is used by move reservations to avoid inverting
95// the move/teardown protocols.
96func freezeAttachmentEffectGatesWith(options attachmentEffectFreezeOptions, clients ...*attachedClient) attachmentTransitionGuard {
97 ordered := orderedAttachmentEffectClients(clients)
98 frozen, acquired := acquireAttachmentEffectGates(ordered, options.done, options.afterFrozen, options.nonblocking)
99 if !acquired {
100 return attachmentTransitionGuard{}
101 }
102
103 interruptByClient := attachmentTransportInterruptsByClient(options.interrupts)
104 frozen.interruptions = make(map[*attachedClient]transportSnapshot, len(interruptByClient))
105 interruptFrozenAttachmentTransports(ordered, interruptByClient)
106 frozen.acquired = true
107 frozen.drained = drainFrozenAttachmentEffects(ordered, interruptByClient, frozen.interruptions, options.done)
108 return frozen
109}
110
111// orderedAttachmentEffectClients snapshots unique participants in immutable lock
112// order. Assigning a previously unused order does not freeze or wait on a gate.
113func orderedAttachmentEffectClients(clients []*attachedClient) []*attachedClient {
114 unique := make(map[*attachedClient]struct{}, len(clients))
115 ordered := make([]*attachedClient, 0, len(clients))
116 for _, ac := range clients {
117 if ac == nil {
118 continue
119 }
120 if _, exists := unique[ac]; exists {
121 continue
122 }
123 unique[ac] = struct{}{}
124 ac.lifecycle.immutableOrder()
125 ordered = append(ordered, ac)
126 }
127 sort.Slice(ordered, func(i, j int) bool {
128 return ordered[i].lifecycle.order.Load() < ordered[j].lifecycle.order.Load()
129 })
130 return ordered
131}
132
133// waitForAttachmentEffectChangeLocked requires g.mu and returns with g.mu held. A
134// true result means the overall deadline won and the caller must stop waiting.
135func waitForAttachmentEffectChangeLocked(g *attachmentLifecycle, done func() <-chan struct{}) bool {
136 if done == nil {
137 g.cond.Wait()
138 return false
139 }
140 changed := g.changed
141 doneCh := done()
142 g.mu.Unlock()
143 select {
144 case <-changed:
145 case <-doneCh:
146 }
147 g.mu.Lock()
148 select {
149 case <-doneCh:
150 return true
151 default:
152 return false
153 }
154}
155
156// acquireAttachmentEffectGates establishes exclusive transition ownership in order.
157// It must run without architecture locks. Deadline rollback releases only this
158// call's partial acquisition and does so in reverse order.
159func acquireAttachmentEffectGates(ordered []*attachedClient, done func() <-chan struct{}, afterFrozen func(*attachedClient), nonblocking bool) (attachmentTransitionGuard, bool) {
160 frozen := attachmentTransitionGuard{clients: make([]*attachedClient, 0, len(ordered))}
161 for _, ac := range ordered {
162 g := &ac.lifecycle
163 g.mu.Lock()
164 g.initLocked()
165 for g.phase == attachmentEffectsFrozen {
166 if nonblocking {
167 g.mu.Unlock()
168 frozen.unfreeze()
169 return attachmentTransitionGuard{}, false
170 }
171 if waitForAttachmentEffectChangeLocked(g, done) {
172 g.mu.Unlock()
173 frozen.unfreeze()
174 return attachmentTransitionGuard{}, false
175 }
176 }
177 g.phase = attachmentEffectsFrozen
178 g.mu.Unlock()
179 frozen.clients = append(frozen.clients, ac)
180 if afterFrozen != nil {
181 afterFrozen(ac)
182 }
183 }
184 return frozen, true
185}
186
187func attachmentTransportInterruptsByClient(interrupts []attachmentTransportInterrupt) map[*attachedClient]transportSnapshot {
188 byClient := make(map[*attachedClient]transportSnapshot, len(interrupts))
189 for _, interrupt := range interrupts {
190 if interrupt.ac != nil && interrupt.transport.transport != nil {
191 byClient[interrupt.ac] = interrupt.transport
192 }
193 }
194 return byClient
195}
196
197// interruptFrozenAttachmentTransports snapshots every exact in-flight link while all
198// gates are frozen, then closes the complete set before any drain wait. One
199// uncooperative Send cannot prevent another participant from being interrupted.
200func interruptFrozenAttachmentTransports(ordered []*attachedClient, interruptByClient map[*attachedClient]transportSnapshot) {
201 toClose := make([]attachmentTransportInterrupt, 0, len(interruptByClient))
202 for _, ac := range ordered {
203 expected, mayInterrupt := interruptByClient[ac]
204 if !mayInterrupt {
205 continue
206 }
207 g := &ac.lifecycle
208 g.mu.Lock()
209 if gateHasTransportEffectLocked(g, expected) {
210 toClose = append(toClose, attachmentTransportInterrupt{ac: ac, transport: expected})
211 }
212 g.mu.Unlock()
213 }
214 for _, interrupt := range toClose {
215 _ = interrupt.ac.closeCapturedTransport(interrupt.transport.transport)
216 }
217}
218
219// drainFrozenAttachmentEffects waits for admissions made before the freeze. A closed
220// deadline permits terminal teardown to continue with the capability frozen;
221// late tickets can then retire without pinning the owner indefinitely.
222func drainFrozenAttachmentEffects(ordered []*attachedClient, interruptByClient, interruptions map[*attachedClient]transportSnapshot, done func() <-chan struct{}) bool {
223 for _, ac := range ordered {
224 g := &ac.lifecycle
225 g.mu.Lock()
226 for g.inFlight != 0 {
227 if waitForAttachmentEffectChangeLocked(g, done) {
228 g.mu.Unlock()
229 return false
230 }
231 }
232 if expected, mayInterrupt := interruptByClient[ac]; mayInterrupt &&
233 g.failedTransport.transport == expected.transport && g.failedTransport.incarnation == expected.incarnation {
234 interruptions[ac] = expected
235 }
236 g.mu.Unlock()
237 }
238 return true
239}
240
241func gateHasTransportEffectLocked(g *attachmentLifecycle, expected transportSnapshot) bool {
242 for _, active := range g.transportEffects {
243 if active.transport == expected.transport && active.incarnation == expected.incarnation {
244 return true
245 }
246 }
247 return false
248}
249
250func (f attachmentTransitionGuard) unfreeze() {
251 for i := len(f.clients) - 1; i >= 0; i-- {
252 g := &f.clients[i].lifecycle
253 g.mu.Lock()
254 g.initLocked()
255 g.phase = attachmentEffectsStable
256 g.signalLocked()
257 g.mu.Unlock()
258 }
259}