fix(remote): commit proxy lifecycle state before publishing it
armProxyWarm cleared the warm token and stopped its timer in phase 1, before phase 2 revalidated the replacement. A lost revalidation therefore left a registered proxy with no timer at all, losing its only path out of the live registry. Reserve the incumbent token instead and bump the lifecycle generation atomically with publication, retiring the old token only once its successor owns that generation. expireWarmProxy published p.warm, p.generation, and p.expired before the registry removal that commits the expiry. Attempt the removal first so a rejected expiry publishes nothing. Also share the canonical d.mu -> sessionCore.mu -> p.mu locking through withProxyLocked and document that ordering at the file level.
- commit
- 845794acf5f65667aa6bc585681be635e3aff6e1
- author
- brice · · 2026-07-31 13:23 UTC
- parent
- 98aa679
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +351 -84
M internal/usecase/daemon/proxy_lifecycle.go +81 -84
| @@ -7,6 +7,13 @@ import ( | |||
| 7 | 7 | "github.com/bnema/vev/internal/ports" | |
| 8 | 8 | ) | |
| 9 | 9 | ||
| 10 | + | // Proxy lifecycle owns the dormant-expiry timer and the link-state publication | |
| 11 | + | // of one remote attachment. It extends the package lock ordering documented at | |
| 12 | + | // the top of client.go: every function in this file acquires | |
| 13 | + | // d.mu -> p.sessionCore.mu -> p.mu, in that order and never any other, and | |
| 14 | + | // holds them only to validate and publish state. Clock, Timer, transport, | |
| 15 | + | // coordinator, and repaint work always runs after all three are released. | |
| 16 | + | ||
| 10 | 17 | const proxyWarmDuration = 5 * time.Minute | |
| 11 | 18 | ||
| 12 | 19 | // proxyWarmTimer is an exact lifecycle capability. Both pointer identity and | |
| @@ -32,6 +39,11 @@ func (t *proxyWarmTimer) stop() { | |||
| 32 | 39 | // armProxyWarm reserves and publishes a fresh five-minute timer only while the | |
| 33 | 40 | // exact proxy remains registered and headless. Clock and Timer methods are | |
| 34 | 41 | // external operations and therefore run outside architecture locks. | |
| 42 | + | // | |
| 43 | + | // The lifecycle generation is bumped and the incumbent token released only in | |
| 44 | + | // phase 2, atomically with publishing the replacement. A failed revalidation | |
| 45 | + | // therefore leaves the already-armed token both installed and current, so the | |
| 46 | + | // proxy can never be left registered without a path out of the live registry. | |
| 35 | 47 | func (d *Daemon) armProxyWarm(p *proxySession) bool { | |
| 36 | 48 | if d == nil || p == nil || d.clock == nil { | |
| 37 | 49 | return false | |
| @@ -48,24 +60,20 @@ func (d *Daemon) armProxyWarm(p *proxySession) bool { | |||
| 48 | 60 | return false | |
| 49 | 61 | } | |
| 50 | 62 | p.mu.Lock() | |
| 51 | - | p.generation++ | |
| 52 | 63 | generation := p.generation | |
| 53 | 64 | old := p.warm | |
| 54 | - | p.warm = nil | |
| 55 | 65 | p.mu.Unlock() | |
| 56 | 66 | p.sessionCore.mu.Unlock() | |
| 57 | 67 | d.mu.Unlock() | |
| 58 | - | old.stop() | |
| 59 | 68 | ||
| 60 | 69 | timer := d.clock.NewTimer(proxyWarmDuration) | |
| 61 | 70 | if timer == nil { | |
| 62 | 71 | return false | |
| 63 | 72 | } | |
| 64 | 73 | token := &proxyWarmTimer{ | |
| 65 | - | generation: generation, | |
| 66 | - | timer: timer, | |
| 67 | - | cancel: make(chan struct{}), | |
| 68 | - | done: make(chan struct{}), | |
| 74 | + | timer: timer, | |
| 75 | + | cancel: make(chan struct{}), | |
| 76 | + | done: make(chan struct{}), | |
| 69 | 77 | } | |
| 70 | 78 | ||
| 71 | 79 | d.mu.Lock() | |
| @@ -75,8 +83,10 @@ func (d *Daemon) armProxyWarm(p *proxySession) bool { | |||
| 75 | 83 | valid = p.client == nil | |
| 76 | 84 | if valid { | |
| 77 | 85 | p.mu.Lock() | |
| 78 | - | valid = p.generation == generation && p.warm == nil | |
| 86 | + | valid = p.generation == generation && p.warm == old | |
| 79 | 87 | if valid { | |
| 88 | + | p.generation++ | |
| 89 | + | token.generation = p.generation | |
| 80 | 90 | p.warm = token | |
| 81 | 91 | } | |
| 82 | 92 | p.mu.Unlock() | |
| @@ -89,6 +99,9 @@ func (d *Daemon) armProxyWarm(p *proxySession) bool { | |||
| 89 | 99 | close(token.done) | |
| 90 | 100 | return false | |
| 91 | 101 | } | |
| 102 | + | // The incumbent token is retired only once its successor owns the exact | |
| 103 | + | // lifecycle generation, never before. | |
| 104 | + | old.stop() | |
| 92 | 105 | ||
| 93 | 106 | go func() { | |
| 94 | 107 | defer close(token.done) | |
| @@ -127,6 +140,15 @@ func (d *Daemon) expireWarmProxy(p *proxySession, token *proxyWarmTimer) bool { | |||
| 127 | 140 | d.mu.Unlock() | |
| 128 | 141 | return false | |
| 129 | 142 | } | |
| 143 | + | // Removal from the registry is what commits this expiry, so it is attempted | |
| 144 | + | // before any terminal lifecycle state is published. unregisterSessionLocked | |
| 145 | + | // only reads d.sessions, which this call already owns; it acquires no lock. | |
| 146 | + | if !d.unregisterSessionLocked(p) { | |
| 147 | + | p.mu.Unlock() | |
| 148 | + | p.sessionCore.mu.Unlock() | |
| 149 | + | d.mu.Unlock() | |
| 150 | + | return false | |
| 151 | + | } | |
| 130 | 152 | p.warm = nil | |
| 131 | 153 | p.generation++ | |
| 132 | 154 | p.expired = true | |
| @@ -134,11 +156,6 @@ func (d *Daemon) expireWarmProxy(p *proxySession, token *proxyWarmTimer) bool { | |||
| 134 | 156 | transport := p.transport | |
| 135 | 157 | coordinator := p.coordinator.Load() | |
| 136 | 158 | p.mu.Unlock() | |
| 137 | - | if !d.unregisterSessionLocked(p) { | |
| 138 | - | p.sessionCore.mu.Unlock() | |
| 139 | - | d.mu.Unlock() | |
| 140 | - | return false | |
| 141 | - | } | |
| 142 | 159 | p.sessionCore.mu.Unlock() | |
| 143 | 160 | empty := len(d.sessions) == 0 | |
| 144 | 161 | if empty { | |
| @@ -163,11 +180,13 @@ func (d *Daemon) expireWarmProxy(p *proxySession, token *proxyWarmTimer) bool { | |||
| 163 | 180 | return true | |
| 164 | 181 | } | |
| 165 | 182 | ||
| 166 | - | // updateProxyLinkState publishes one transport callback only if the exact proxy, | |
| 167 | - | // link generation, and transport remain current. Repaint is deliberately after | |
| 168 | - | // all architecture locks have been released. | |
| 169 | - | func (d *Daemon) updateProxyLinkState(p *proxySession, generation uint64, transport ports.Transport, state ports.LinkState) bool { | |
| 170 | - | if d == nil || p == nil || transport == nil { | |
| 183 | + | // withProxyLocked evaluates one lifecycle predicate under the canonical | |
| 184 | + | // d.mu -> p.sessionCore.mu -> p.mu order, but only while p is still the exact | |
| 185 | + | // registered session. update owns every state mutation it wants published and | |
| 186 | + | // reports whether it applied. Repaint is deliberately after all architecture | |
| 187 | + | // locks have been released. | |
| 188 | + | func (d *Daemon) withProxyLocked(p *proxySession, update func() bool) bool { | |
| 189 | + | if d == nil || p == nil { | |
| 171 | 190 | return false | |
| 172 | 191 | } | |
| 173 | 192 | d.mu.Lock() | |
| @@ -177,10 +196,7 @@ func (d *Daemon) updateProxyLinkState(p *proxySession, generation uint64, transp | |||
| 177 | 196 | } | |
| 178 | 197 | p.sessionCore.mu.Lock() | |
| 179 | 198 | p.mu.Lock() | |
| 180 | - | current := !p.expired && p.linkGeneration == generation && p.transport == transport | |
| 181 | - | if current { | |
| 182 | - | p.linkState = state | |
| 183 | - | } | |
| 199 | + | current := update() | |
| 184 | 200 | ac := p.client | |
| 185 | 201 | p.mu.Unlock() | |
| 186 | 202 | p.sessionCore.mu.Unlock() | |
| @@ -191,52 +207,40 @@ func (d *Daemon) updateProxyLinkState(p *proxySession, generation uint64, transp | |||
| 191 | 207 | return current | |
| 192 | 208 | } | |
| 193 | 209 | ||
| 210 | + | // updateProxyLinkState publishes one transport callback only if the exact proxy, | |
| 211 | + | // link generation, and transport remain current. | |
| 212 | + | func (d *Daemon) updateProxyLinkState(p *proxySession, generation uint64, transport ports.Transport, state ports.LinkState) bool { | |
| 213 | + | if d == nil || p == nil || transport == nil { | |
| 214 | + | return false | |
| 215 | + | } | |
| 216 | + | return d.withProxyLocked(p, func() bool { | |
| 217 | + | current := !p.expired && p.linkGeneration == generation && p.transport == transport | |
| 218 | + | if current { | |
| 219 | + | p.linkState = state | |
| 220 | + | } | |
| 221 | + | return current | |
| 222 | + | }) | |
| 223 | + | } | |
| 224 | + | ||
| 194 | 225 | // updateProxyDisconnectedState publishes a state after the exact generation's | |
| 195 | 226 | // transport has retired. It is used while resume dialing is pending or failed. | |
| 196 | 227 | func (d *Daemon) updateProxyDisconnectedState(p *proxySession, generation uint64, state ports.LinkState) bool { | |
| 197 | 228 | if d == nil || p == nil { | |
| 198 | 229 | return false | |
| 199 | 230 | } | |
| 200 | - | d.mu.Lock() | |
| 201 | - | if d.sessions[p.id] != p { | |
| 202 | - | d.mu.Unlock() | |
| 203 | - | return false | |
| 204 | - | } | |
| 205 | - | p.sessionCore.mu.Lock() | |
| 206 | - | p.mu.Lock() | |
| 207 | - | current := !p.expired && p.linkGeneration == generation && p.transport == nil | |
| 208 | - | if current { | |
| 209 | - | p.linkState = state | |
| 210 | - | } | |
| 211 | - | ac := p.client | |
| 212 | - | p.mu.Unlock() | |
| 213 | - | p.sessionCore.mu.Unlock() | |
| 214 | - | d.mu.Unlock() | |
| 215 | - | if current && ac != nil { | |
| 216 | - | d.invalidateRender(p, ac, false, "proxy_lifecycle.go") | |
| 217 | - | } | |
| 218 | - | return current | |
| 231 | + | return d.withProxyLocked(p, func() bool { | |
| 232 | + | current := !p.expired && p.linkGeneration == generation && p.transport == nil | |
| 233 | + | if current { | |
| 234 | + | p.linkState = state | |
| 235 | + | } | |
| 236 | + | return current | |
| 237 | + | }) | |
| 219 | 238 | } | |
| 220 | 239 | ||
| 221 | 240 | func (d *Daemon) repaintProxyLifecycle(p *proxySession) { | |
| 222 | - | if d == nil || p == nil { | |
| 223 | - | return | |
| 224 | - | } | |
| 225 | - | d.mu.Lock() | |
| 226 | - | if d.sessions[p.id] != p { | |
| 227 | - | d.mu.Unlock() | |
| 228 | - | return | |
| 229 | - | } | |
| 230 | - | p.sessionCore.mu.Lock() | |
| 231 | - | p.mu.Lock() | |
| 232 | - | current := !p.expired && p.transport != nil | |
| 233 | - | ac := p.client | |
| 234 | - | p.mu.Unlock() | |
| 235 | - | p.sessionCore.mu.Unlock() | |
| 236 | - | d.mu.Unlock() | |
| 237 | - | if current && ac != nil { | |
| 238 | - | d.invalidateRender(p, ac, false, "proxy_lifecycle.go") | |
| 239 | - | } | |
| 241 | + | d.withProxyLocked(p, func() bool { | |
| 242 | + | return !p.expired && p.transport != nil | |
| 243 | + | }) | |
| 240 | 244 | } | |
| 241 | 245 | ||
| 242 | 246 | // markProxyReplaced makes ReasonReplaced terminal for this exact remote link. | |
| @@ -249,35 +253,28 @@ func (d *Daemon) markProxyReplaced(p *proxySession, generation uint64, transport | |||
| 249 | 253 | if d == nil || p == nil || transport == nil { | |
| 250 | 254 | return false | |
| 251 | 255 | } | |
| 252 | - | d.mu.Lock() | |
| 253 | - | if d.sessions[p.id] != p { | |
| 254 | - | d.mu.Unlock() | |
| 255 | - | return false | |
| 256 | - | } | |
| 257 | - | p.sessionCore.mu.Lock() | |
| 258 | - | p.mu.Lock() | |
| 259 | - | current := p.linkGeneration == generation && p.transport == transport | |
| 260 | - | if current { | |
| 261 | - | p.expired = true | |
| 262 | - | p.linkState = ports.LinkStateDead | |
| 263 | - | p.generation++ | |
| 264 | - | if p.warm != nil { | |
| 265 | - | // Rekey the already-published exact timer instead of stopping it. This | |
| 266 | - | // preserves its deadline while invalidating older lifecycle observers. | |
| 267 | - | p.warm.generation = p.generation | |
| 256 | + | // Arming and repainting are mutually exclusive: needsWarm requires a headless | |
| 257 | + | // proxy, and the helper only repaints an attached one. Deferring the arm until | |
| 258 | + | // after the helper's repaint therefore preserves the published order. | |
| 259 | + | var needsWarm bool | |
| 260 | + | current := d.withProxyLocked(p, func() bool { | |
| 261 | + | current := p.linkGeneration == generation && p.transport == transport | |
| 262 | + | if current { | |
| 263 | + | p.expired = true | |
| 264 | + | p.linkState = ports.LinkStateDead | |
| 265 | + | p.generation++ | |
| 266 | + | if p.warm != nil { | |
| 267 | + | // Rekey the already-published exact timer instead of stopping it. This | |
| 268 | + | // preserves its deadline while invalidating older lifecycle observers. | |
| 269 | + | p.warm.generation = p.generation | |
| 270 | + | } | |
| 268 | 271 | } | |
| 269 | - | } | |
| 270 | - | ac := p.client | |
| 271 | - | needsWarm := current && ac == nil && p.warm == nil | |
| 272 | - | p.mu.Unlock() | |
| 273 | - | p.sessionCore.mu.Unlock() | |
| 274 | - | d.mu.Unlock() | |
| 272 | + | needsWarm = current && p.client == nil && p.warm == nil | |
| 273 | + | return current | |
| 274 | + | }) | |
| 275 | 275 | if needsWarm { | |
| 276 | 276 | d.armProxyWarm(p) | |
| 277 | 277 | } | |
| 278 | - | if current && ac != nil { | |
| 279 | - | d.invalidateRender(p, ac, false, "proxy_lifecycle.go") | |
| 280 | - | } | |
| 281 | 278 | return current | |
| 282 | 279 | } | |
| 283 | 280 | ||
| @@ -7,6 +7,13 @@ import ( | |||||
| 7 | "github.com/bnema/vev/internal/ports" | 7 | "github.com/bnema/vev/internal/ports" | ||
| 8 | ) | 8 | ) | ||
| 9 | 9 | ||||
| 10 | + | // Proxy lifecycle owns the dormant-expiry timer and the link-state publication | |||
| 11 | + | // of one remote attachment. It extends the package lock ordering documented at | |||
| 12 | + | // the top of client.go: every function in this file acquires | |||
| 13 | + | // d.mu -> p.sessionCore.mu -> p.mu, in that order and never any other, and | |||
| 14 | + | // holds them only to validate and publish state. Clock, Timer, transport, | |||
| 15 | + | // coordinator, and repaint work always runs after all three are released. | |||
| 16 | + | ||||
| 10 | const proxyWarmDuration = 5 * time.Minute | 17 | const proxyWarmDuration = 5 * time.Minute | ||
| 11 | 18 | ||||
| 12 | // proxyWarmTimer is an exact lifecycle capability. Both pointer identity and | 19 | // proxyWarmTimer is an exact lifecycle capability. Both pointer identity and | ||
| @@ -32,6 +39,11 @@ func (t *proxyWarmTimer) stop() { | |||||
| 32 | // armProxyWarm reserves and publishes a fresh five-minute timer only while the | 39 | // armProxyWarm reserves and publishes a fresh five-minute timer only while the | ||
| 33 | // exact proxy remains registered and headless. Clock and Timer methods are | 40 | // exact proxy remains registered and headless. Clock and Timer methods are | ||
| 34 | // external operations and therefore run outside architecture locks. | 41 | // external operations and therefore run outside architecture locks. | ||
| 42 | + | // | |||
| 43 | + | // The lifecycle generation is bumped and the incumbent token released only in | |||
| 44 | + | // phase 2, atomically with publishing the replacement. A failed revalidation | |||
| 45 | + | // therefore leaves the already-armed token both installed and current, so the | |||
| 46 | + | // proxy can never be left registered without a path out of the live registry. | |||
| 35 | func (d *Daemon) armProxyWarm(p *proxySession) bool { | 47 | func (d *Daemon) armProxyWarm(p *proxySession) bool { | ||
| 36 | if d == nil || p == nil || d.clock == nil { | 48 | if d == nil || p == nil || d.clock == nil { | ||
| 37 | return false | 49 | return false | ||
| @@ -48,24 +60,20 @@ func (d *Daemon) armProxyWarm(p *proxySession) bool { | |||||
| 48 | return false | 60 | return false | ||
| 49 | } | 61 | } | ||
| 50 | p.mu.Lock() | 62 | p.mu.Lock() | ||
| 51 | - | p.generation++ | |||
| 52 | generation := p.generation | 63 | generation := p.generation | ||
| 53 | old := p.warm | 64 | old := p.warm | ||
| 54 | - | p.warm = nil | |||
| 55 | p.mu.Unlock() | 65 | p.mu.Unlock() | ||
| 56 | p.sessionCore.mu.Unlock() | 66 | p.sessionCore.mu.Unlock() | ||
| 57 | d.mu.Unlock() | 67 | d.mu.Unlock() | ||
| 58 | - | old.stop() | |||
| 59 | 68 | ||||
| 60 | timer := d.clock.NewTimer(proxyWarmDuration) | 69 | timer := d.clock.NewTimer(proxyWarmDuration) | ||
| 61 | if timer == nil { | 70 | if timer == nil { | ||
| 62 | return false | 71 | return false | ||
| 63 | } | 72 | } | ||
| 64 | token := &proxyWarmTimer{ | 73 | token := &proxyWarmTimer{ | ||
| 65 | - | generation: generation, | 74 | + | timer: timer, |
| 66 | - | timer: timer, | 75 | + | cancel: make(chan struct{}), |
| 67 | - | cancel: make(chan struct{}), | 76 | + | done: make(chan struct{}), |
| 68 | - | done: make(chan struct{}), | |||
| 69 | } | 77 | } | ||
| 70 | 78 | ||||
| 71 | d.mu.Lock() | 79 | d.mu.Lock() | ||
| @@ -75,8 +83,10 @@ func (d *Daemon) armProxyWarm(p *proxySession) bool { | |||||
| 75 | valid = p.client == nil | 83 | valid = p.client == nil | ||
| 76 | if valid { | 84 | if valid { | ||
| 77 | p.mu.Lock() | 85 | p.mu.Lock() | ||
| 78 | - | valid = p.generation == generation && p.warm == nil | 86 | + | valid = p.generation == generation && p.warm == old |
| 79 | if valid { | 87 | if valid { | ||
| 88 | + | p.generation++ | |||
| 89 | + | token.generation = p.generation | |||
| 80 | p.warm = token | 90 | p.warm = token | ||
| 81 | } | 91 | } | ||
| 82 | p.mu.Unlock() | 92 | p.mu.Unlock() | ||
| @@ -89,6 +99,9 @@ func (d *Daemon) armProxyWarm(p *proxySession) bool { | |||||
| 89 | close(token.done) | 99 | close(token.done) | ||
| 90 | return false | 100 | return false | ||
| 91 | } | 101 | } | ||
| 102 | + | // The incumbent token is retired only once its successor owns the exact | |||
| 103 | + | // lifecycle generation, never before. | |||
| 104 | + | old.stop() | |||
| 92 | 105 | ||||
| 93 | go func() { | 106 | go func() { | ||
| 94 | defer close(token.done) | 107 | defer close(token.done) | ||
| @@ -127,6 +140,15 @@ func (d *Daemon) expireWarmProxy(p *proxySession, token *proxyWarmTimer) bool { | |||||
| 127 | d.mu.Unlock() | 140 | d.mu.Unlock() | ||
| 128 | return false | 141 | return false | ||
| 129 | } | 142 | } | ||
| 143 | + | // Removal from the registry is what commits this expiry, so it is attempted | |||
| 144 | + | // before any terminal lifecycle state is published. unregisterSessionLocked | |||
| 145 | + | // only reads d.sessions, which this call already owns; it acquires no lock. | |||
| 146 | + | if !d.unregisterSessionLocked(p) { | |||
| 147 | + | p.mu.Unlock() | |||
| 148 | + | p.sessionCore.mu.Unlock() | |||
| 149 | + | d.mu.Unlock() | |||
| 150 | + | return false | |||
| 151 | + | } | |||
| 130 | p.warm = nil | 152 | p.warm = nil | ||
| 131 | p.generation++ | 153 | p.generation++ | ||
| 132 | p.expired = true | 154 | p.expired = true | ||
| @@ -134,11 +156,6 @@ func (d *Daemon) expireWarmProxy(p *proxySession, token *proxyWarmTimer) bool { | |||||
| 134 | transport := p.transport | 156 | transport := p.transport | ||
| 135 | coordinator := p.coordinator.Load() | 157 | coordinator := p.coordinator.Load() | ||
| 136 | p.mu.Unlock() | 158 | p.mu.Unlock() | ||
| 137 | - | if !d.unregisterSessionLocked(p) { | |||
| 138 | - | p.sessionCore.mu.Unlock() | |||
| 139 | - | d.mu.Unlock() | |||
| 140 | - | return false | |||
| 141 | - | } | |||
| 142 | p.sessionCore.mu.Unlock() | 159 | p.sessionCore.mu.Unlock() | ||
| 143 | empty := len(d.sessions) == 0 | 160 | empty := len(d.sessions) == 0 | ||
| 144 | if empty { | 161 | if empty { | ||
| @@ -163,11 +180,13 @@ func (d *Daemon) expireWarmProxy(p *proxySession, token *proxyWarmTimer) bool { | |||||
| 163 | return true | 180 | return true | ||
| 164 | } | 181 | } | ||
| 165 | 182 | ||||
| 166 | - | // updateProxyLinkState publishes one transport callback only if the exact proxy, | 183 | + | // withProxyLocked evaluates one lifecycle predicate under the canonical |
| 167 | - | // link generation, and transport remain current. Repaint is deliberately after | 184 | + | // d.mu -> p.sessionCore.mu -> p.mu order, but only while p is still the exact |
| 168 | - | // all architecture locks have been released. | 185 | + | // registered session. update owns every state mutation it wants published and |
| 169 | - | func (d *Daemon) updateProxyLinkState(p *proxySession, generation uint64, transport ports.Transport, state ports.LinkState) bool { | 186 | + | // reports whether it applied. Repaint is deliberately after all architecture |
| 170 | - | if d == nil || p == nil || transport == nil { | 187 | + | // locks have been released. |
| 188 | + | func (d *Daemon) withProxyLocked(p *proxySession, update func() bool) bool { | |||
| 189 | + | if d == nil || p == nil { | |||
| 171 | return false | 190 | return false | ||
| 172 | } | 191 | } | ||
| 173 | d.mu.Lock() | 192 | d.mu.Lock() | ||
| @@ -177,10 +196,7 @@ func (d *Daemon) updateProxyLinkState(p *proxySession, generation uint64, transp | |||||
| 177 | } | 196 | } | ||
| 178 | p.sessionCore.mu.Lock() | 197 | p.sessionCore.mu.Lock() | ||
| 179 | p.mu.Lock() | 198 | p.mu.Lock() | ||
| 180 | - | current := !p.expired && p.linkGeneration == generation && p.transport == transport | 199 | + | current := update() |
| 181 | - | if current { | |||
| 182 | - | p.linkState = state | |||
| 183 | - | } | |||
| 184 | ac := p.client | 200 | ac := p.client | ||
| 185 | p.mu.Unlock() | 201 | p.mu.Unlock() | ||
| 186 | p.sessionCore.mu.Unlock() | 202 | p.sessionCore.mu.Unlock() | ||
| @@ -191,52 +207,40 @@ func (d *Daemon) updateProxyLinkState(p *proxySession, generation uint64, transp | |||||
| 191 | return current | 207 | return current | ||
| 192 | } | 208 | } | ||
| 193 | 209 | ||||
| 210 | + | // updateProxyLinkState publishes one transport callback only if the exact proxy, | |||
| 211 | + | // link generation, and transport remain current. | |||
| 212 | + | func (d *Daemon) updateProxyLinkState(p *proxySession, generation uint64, transport ports.Transport, state ports.LinkState) bool { | |||
| 213 | + | if d == nil || p == nil || transport == nil { | |||
| 214 | + | return false | |||
| 215 | + | } | |||
| 216 | + | return d.withProxyLocked(p, func() bool { | |||
| 217 | + | current := !p.expired && p.linkGeneration == generation && p.transport == transport | |||
| 218 | + | if current { | |||
| 219 | + | p.linkState = state | |||
| 220 | + | } | |||
| 221 | + | return current | |||
| 222 | + | }) | |||
| 223 | + | } | |||
| 224 | + | ||||
| 194 | // updateProxyDisconnectedState publishes a state after the exact generation's | 225 | // updateProxyDisconnectedState publishes a state after the exact generation's | ||
| 195 | // transport has retired. It is used while resume dialing is pending or failed. | 226 | // transport has retired. It is used while resume dialing is pending or failed. | ||
| 196 | func (d *Daemon) updateProxyDisconnectedState(p *proxySession, generation uint64, state ports.LinkState) bool { | 227 | func (d *Daemon) updateProxyDisconnectedState(p *proxySession, generation uint64, state ports.LinkState) bool { | ||
| 197 | if d == nil || p == nil { | 228 | if d == nil || p == nil { | ||
| 198 | return false | 229 | return false | ||
| 199 | } | 230 | } | ||
| 200 | - | d.mu.Lock() | 231 | + | return d.withProxyLocked(p, func() bool { |
| 201 | - | if d.sessions[p.id] != p { | 232 | + | current := !p.expired && p.linkGeneration == generation && p.transport == nil |
| 202 | - | d.mu.Unlock() | 233 | + | if current { |
| 203 | - | return false | 234 | + | p.linkState = state |
| 204 | - | } | 235 | + | } |
| 205 | - | p.sessionCore.mu.Lock() | 236 | + | return current |
| 206 | - | p.mu.Lock() | 237 | + | }) |
| 207 | - | current := !p.expired && p.linkGeneration == generation && p.transport == nil | |||
| 208 | - | if current { | |||
| 209 | - | p.linkState = state | |||
| 210 | - | } | |||
| 211 | - | ac := p.client | |||
| 212 | - | p.mu.Unlock() | |||
| 213 | - | p.sessionCore.mu.Unlock() | |||
| 214 | - | d.mu.Unlock() | |||
| 215 | - | if current && ac != nil { | |||
| 216 | - | d.invalidateRender(p, ac, false, "proxy_lifecycle.go") | |||
| 217 | - | } | |||
| 218 | - | return current | |||
| 219 | } | 238 | } | ||
| 220 | 239 | ||||
| 221 | func (d *Daemon) repaintProxyLifecycle(p *proxySession) { | 240 | func (d *Daemon) repaintProxyLifecycle(p *proxySession) { | ||
| 222 | - | if d == nil || p == nil { | 241 | + | d.withProxyLocked(p, func() bool { |
| 223 | - | return | 242 | + | return !p.expired && p.transport != nil |
| 224 | - | } | 243 | + | }) |
| 225 | - | d.mu.Lock() | |||
| 226 | - | if d.sessions[p.id] != p { | |||
| 227 | - | d.mu.Unlock() | |||
| 228 | - | return | |||
| 229 | - | } | |||
| 230 | - | p.sessionCore.mu.Lock() | |||
| 231 | - | p.mu.Lock() | |||
| 232 | - | current := !p.expired && p.transport != nil | |||
| 233 | - | ac := p.client | |||
| 234 | - | p.mu.Unlock() | |||
| 235 | - | p.sessionCore.mu.Unlock() | |||
| 236 | - | d.mu.Unlock() | |||
| 237 | - | if current && ac != nil { | |||
| 238 | - | d.invalidateRender(p, ac, false, "proxy_lifecycle.go") | |||
| 239 | - | } | |||
| 240 | } | 244 | } | ||
| 241 | 245 | ||||
| 242 | // markProxyReplaced makes ReasonReplaced terminal for this exact remote link. | 246 | // markProxyReplaced makes ReasonReplaced terminal for this exact remote link. | ||
| @@ -249,35 +253,28 @@ func (d *Daemon) markProxyReplaced(p *proxySession, generation uint64, transport | |||||
| 249 | if d == nil || p == nil || transport == nil { | 253 | if d == nil || p == nil || transport == nil { | ||
| 250 | return false | 254 | return false | ||
| 251 | } | 255 | } | ||
| 252 | - | d.mu.Lock() | 256 | + | // Arming and repainting are mutually exclusive: needsWarm requires a headless |
| 253 | - | if d.sessions[p.id] != p { | 257 | + | // proxy, and the helper only repaints an attached one. Deferring the arm until |
| 254 | - | d.mu.Unlock() | 258 | + | // after the helper's repaint therefore preserves the published order. |
| 255 | - | return false | 259 | + | var needsWarm bool |
| 256 | - | } | 260 | + | current := d.withProxyLocked(p, func() bool { |
| 257 | - | p.sessionCore.mu.Lock() | 261 | + | current := p.linkGeneration == generation && p.transport == transport |
| 258 | - | p.mu.Lock() | 262 | + | if current { |
| 259 | - | current := p.linkGeneration == generation && p.transport == transport | 263 | + | p.expired = true |
| 260 | - | if current { | 264 | + | p.linkState = ports.LinkStateDead |
| 261 | - | p.expired = true | 265 | + | p.generation++ |
| 262 | - | p.linkState = ports.LinkStateDead | 266 | + | if p.warm != nil { |
| 263 | - | p.generation++ | 267 | + | // Rekey the already-published exact timer instead of stopping it. This |
| 264 | - | if p.warm != nil { | 268 | + | // preserves its deadline while invalidating older lifecycle observers. |
| 265 | - | // Rekey the already-published exact timer instead of stopping it. This | 269 | + | p.warm.generation = p.generation |
| 266 | - | // preserves its deadline while invalidating older lifecycle observers. | 270 | + | } |
| 267 | - | p.warm.generation = p.generation | |||
| 268 | } | 271 | } | ||
| 269 | - | } | 272 | + | needsWarm = current && p.client == nil && p.warm == nil |
| 270 | - | ac := p.client | 273 | + | return current |
| 271 | - | needsWarm := current && ac == nil && p.warm == nil | 274 | + | }) |
| 272 | - | p.mu.Unlock() | |||
| 273 | - | p.sessionCore.mu.Unlock() | |||
| 274 | - | d.mu.Unlock() | |||
| 275 | if needsWarm { | 275 | if needsWarm { | ||
| 276 | d.armProxyWarm(p) | 276 | d.armProxyWarm(p) | ||
| 277 | } | 277 | } | ||
| 278 | - | if current && ac != nil { | |||
| 279 | - | d.invalidateRender(p, ac, false, "proxy_lifecycle.go") | |||
| 280 | - | } | |||
| 281 | return current | 278 | return current | ||
| 282 | } | 279 | } | ||
| 283 | 280 | ||||
A internal/usecase/daemon/proxy_lifecycle_test.go +270 -0
| @@ -0,0 +1,270 @@ | |||
| 1 | + | package daemon | |
| 2 | + | ||
| 3 | + | import ( | |
| 4 | + | "sync" | |
| 5 | + | "testing" | |
| 6 | + | "time" | |
| 7 | + | ||
| 8 | + | "github.com/bnema/vev/internal/domain" | |
| 9 | + | "github.com/bnema/vev/internal/ports" | |
| 10 | + | "github.com/stretchr/testify/require" | |
| 11 | + | ) | |
| 12 | + | ||
| 13 | + | // warmHookClock runs a one-shot hook while armProxyWarm sits between its | |
| 14 | + | // reservation and its publication, so a lifecycle race can be interleaved | |
| 15 | + | // deterministically instead of being waited for. | |
| 16 | + | type warmHookClock struct { | |
| 17 | + | timers chan *signalTimer | |
| 18 | + | ||
| 19 | + | mu sync.Mutex | |
| 20 | + | hook func() | |
| 21 | + | } | |
| 22 | + | ||
| 23 | + | func (c *warmHookClock) Now() time.Time { return time.Time{} } | |
| 24 | + | ||
| 25 | + | func (c *warmHookClock) NewTimer(d time.Duration) ports.Timer { | |
| 26 | + | timer := &signalTimer{ch: make(chan time.Time, 1), duration: d} | |
| 27 | + | c.mu.Lock() | |
| 28 | + | hook := c.hook | |
| 29 | + | c.hook = nil | |
| 30 | + | c.mu.Unlock() | |
| 31 | + | if hook != nil { | |
| 32 | + | hook() | |
| 33 | + | } | |
| 34 | + | if c.timers != nil { | |
| 35 | + | c.timers <- timer | |
| 36 | + | } | |
| 37 | + | return timer | |
| 38 | + | } | |
| 39 | + | ||
| 40 | + | func (c *warmHookClock) setHook(hook func()) { | |
| 41 | + | c.mu.Lock() | |
| 42 | + | c.hook = hook | |
| 43 | + | c.mu.Unlock() | |
| 44 | + | } | |
| 45 | + | ||
| 46 | + | func registerLifecycleProxy(t *testing.T, d *Daemon, key domain.RemoteSessionKey) *proxySession { | |
| 47 | + | t.Helper() | |
| 48 | + | p, err := newProxySession(key, domain.Size{Cols: 80, Rows: 24}) | |
| 49 | + | require.NoError(t, err) | |
| 50 | + | d.mu.Lock() | |
| 51 | + | registered := d.registerSessionLocked(p) | |
| 52 | + | d.mu.Unlock() | |
| 53 | + | require.True(t, registered, "lifecycle fixture proxy was not published") | |
| 54 | + | return p | |
| 55 | + | } | |
| 56 | + | ||
| 57 | + | func newProxyLifecycleFixture(t *testing.T) (*Daemon, *proxySession, *warmHookClock) { | |
| 58 | + | t.Helper() | |
| 59 | + | clock := &warmHookClock{timers: make(chan *signalTimer, 4)} | |
| 60 | + | d := newProxyTestDaemon(t, nil, clock) | |
| 61 | + | return d, registerLifecycleProxy(t, d, domain.RemoteSessionKey{Host: "arch", Name: "work"}), clock | |
| 62 | + | } | |
| 63 | + | ||
| 64 | + | func proxyWarmToken(p *proxySession) (*proxyWarmTimer, uint64) { | |
| 65 | + | p.mu.Lock() | |
| 66 | + | defer p.mu.Unlock() | |
| 67 | + | return p.warm, p.generation | |
| 68 | + | } | |
| 69 | + | ||
| 70 | + | func setProxyLifecycleClient(p *proxySession, ac *attachedClient) { | |
| 71 | + | p.sessionCore.mu.Lock() | |
| 72 | + | p.client = ac | |
| 73 | + | p.sessionCore.mu.Unlock() | |
| 74 | + | } | |
| 75 | + | ||
| 76 | + | func proxyRegistered(d *Daemon, p *proxySession) bool { | |
| 77 | + | d.mu.Lock() | |
| 78 | + | defer d.mu.Unlock() | |
| 79 | + | return d.sessions[p.id] == p | |
| 80 | + | } | |
| 81 | + | ||
| 82 | + | func proxyExpired(p *proxySession) bool { | |
| 83 | + | p.mu.Lock() | |
| 84 | + | defer p.mu.Unlock() | |
| 85 | + | return p.expired | |
| 86 | + | } | |
| 87 | + | ||
| 88 | + | func armWarmTimer(t *testing.T, d *Daemon, p *proxySession, clock *warmHookClock) (*proxyWarmTimer, *signalTimer) { | |
| 89 | + | t.Helper() | |
| 90 | + | require.True(t, d.armProxyWarm(p), "warm timer was not published") | |
| 91 | + | timer := awaitTestValue(t, clock.timers, "warm timer was never created") | |
| 92 | + | token, _ := proxyWarmToken(p) | |
| 93 | + | require.NotNil(t, token, "published arm left no warm token") | |
| 94 | + | return token, timer | |
| 95 | + | } | |
| 96 | + | ||
| 97 | + | // TestArmProxyWarmKeepsExpiryPathWhenPublicationFails pins the two-phase arm: | |
| 98 | + | // a re-arm that loses its revalidation must leave the incumbent timer both | |
| 99 | + | // installed and current, otherwise a registered proxy is left with no path out | |
| 100 | + | // of the live registry. | |
| 101 | + | func TestArmProxyWarmKeepsExpiryPathWhenPublicationFails(t *testing.T) { | |
| 102 | + | tests := []struct { | |
| 103 | + | name string | |
| 104 | + | race func(d *Daemon, p *proxySession) | |
| 105 | + | undo func(d *Daemon, p *proxySession) | |
| 106 | + | }{ | |
| 107 | + | { | |
| 108 | + | name: "client attaches during arm", | |
| 109 | + | race: func(_ *Daemon, p *proxySession) { setProxyLifecycleClient(p, &attachedClient{}) }, | |
| 110 | + | undo: func(_ *Daemon, p *proxySession) { setProxyLifecycleClient(p, nil) }, | |
| 111 | + | }, | |
| 112 | + | { | |
| 113 | + | name: "replacement rekeys the lifecycle generation", | |
| 114 | + | race: func(_ *Daemon, p *proxySession) { | |
| 115 | + | p.mu.Lock() | |
| 116 | + | p.expired = true | |
| 117 | + | p.generation++ | |
| 118 | + | if p.warm != nil { | |
| 119 | + | p.warm.generation = p.generation | |
| 120 | + | } | |
| 121 | + | p.mu.Unlock() | |
| 122 | + | }, | |
| 123 | + | }, | |
| 124 | + | { | |
| 125 | + | name: "daemon starts closing during arm", | |
| 126 | + | race: func(d *Daemon, _ *proxySession) { | |
| 127 | + | d.mu.Lock() | |
| 128 | + | d.closing = true | |
| 129 | + | d.mu.Unlock() | |
| 130 | + | }, | |
| 131 | + | undo: func(d *Daemon, _ *proxySession) { | |
| 132 | + | d.mu.Lock() | |
| 133 | + | d.closing = false | |
| 134 | + | d.mu.Unlock() | |
| 135 | + | }, | |
| 136 | + | }, | |
| 137 | + | } | |
| 138 | + | for _, tt := range tests { | |
| 139 | + | t.Run(tt.name, func(t *testing.T) { | |
| 140 | + | d, p, clock := newProxyLifecycleFixture(t) | |
| 141 | + | token, timer := armWarmTimer(t, d, p, clock) | |
| 142 | + | ||
| 143 | + | clock.setHook(func() { tt.race(d, p) }) | |
| 144 | + | require.False(t, d.armProxyWarm(p), "raced re-arm must not publish a replacement") | |
| 145 | + | awaitTestValue(t, clock.timers, "raced re-arm never reached its clock") | |
| 146 | + | ||
| 147 | + | retained, generation := proxyWarmToken(p) | |
| 148 | + | require.NotNil(t, retained, "failed re-arm stripped the proxy of its only expiry path") | |
| 149 | + | require.Same(t, token, retained, "failed re-arm replaced the armed expiry timer") | |
| 150 | + | require.Equal(t, generation, retained.generation, "retained timer no longer owns the lifecycle generation") | |
| 151 | + | require.True(t, proxyRegistered(d, p), "a failed re-arm must not unregister the proxy") | |
| 152 | + | ||
| 153 | + | if tt.undo != nil { | |
| 154 | + | tt.undo(d, p) | |
| 155 | + | } | |
| 156 | + | timer.ch <- time.Time{} | |
| 157 | + | awaitTestCompletion(t, token.done, "retained warm timer never ran to completion") | |
| 158 | + | require.False(t, proxyRegistered(d, p), "retained warm timer did not expire the dormant proxy") | |
| 159 | + | require.True(t, proxyExpired(p), "expiry did not publish terminal lifecycle state") | |
| 160 | + | awaitTestCompletion(t, d.done, "last proxy expiry did not complete the daemon") | |
| 161 | + | }) | |
| 162 | + | } | |
| 163 | + | } | |
| 164 | + | ||
| 165 | + | func TestMarkProxyReplacedWarmTimerOwnership(t *testing.T) { | |
| 166 | + | t.Run("rekeys an already armed timer", func(t *testing.T) { | |
| 167 | + | d, p, clock := newProxyLifecycleFixture(t) | |
| 168 | + | transport := newProxyTestTransport() | |
| 169 | + | t.Cleanup(func() { _ = transport.Close() }) | |
| 170 | + | generation, _ := p.installTransport(transport) | |
| 171 | + | token, timer := armWarmTimer(t, d, p, clock) | |
| 172 | + | ||
| 173 | + | require.True(t, d.markProxyReplaced(p, generation, transport)) | |
| 174 | + | require.Empty(t, clock.timers, "replacement armed a second timer instead of rekeying") | |
| 175 | + | ||
| 176 | + | rekeyed, lifecycle := proxyWarmToken(p) | |
| 177 | + | require.Same(t, token, rekeyed, "replacement replaced the armed timer") | |
| 178 | + | require.Equal(t, lifecycle, rekeyed.generation, "replacement did not rekey the armed timer") | |
| 179 | + | require.True(t, proxyExpired(p)) | |
| 180 | + | p.mu.Lock() | |
| 181 | + | state := p.linkState | |
| 182 | + | p.mu.Unlock() | |
| 183 | + | require.Equal(t, ports.LinkStateDead, state) | |
| 184 | + | ||
| 185 | + | timer.ch <- time.Time{} | |
| 186 | + | awaitTestCompletion(t, token.done, "rekeyed warm timer never ran to completion") | |
| 187 | + | require.False(t, proxyRegistered(d, p), "rekeyed warm timer did not expire the proxy") | |
| 188 | + | }) | |
| 189 | + | ||
| 190 | + | t.Run("arms an expiry for a headless proxy", func(t *testing.T) { | |
| 191 | + | d, p, clock := newProxyLifecycleFixture(t) | |
| 192 | + | transport := newProxyTestTransport() | |
| 193 | + | t.Cleanup(func() { _ = transport.Close() }) | |
| 194 | + | generation, _ := p.installTransport(transport) | |
| 195 | + | ||
| 196 | + | require.True(t, d.markProxyReplaced(p, generation, transport)) | |
| 197 | + | timer := awaitTestValue(t, clock.timers, "replacement did not arm a warm timer") | |
| 198 | + | token, lifecycle := proxyWarmToken(p) | |
| 199 | + | require.NotNil(t, token, "replaced headless proxy has no expiry path") | |
| 200 | + | require.Equal(t, lifecycle, token.generation) | |
| 201 | + | ||
| 202 | + | timer.ch <- time.Time{} | |
| 203 | + | awaitTestCompletion(t, token.done, "warm timer never ran to completion") | |
| 204 | + | require.False(t, proxyRegistered(d, p)) | |
| 205 | + | }) | |
| 206 | + | ||
| 207 | + | t.Run("stale link generation publishes nothing", func(t *testing.T) { | |
| 208 | + | d, p, _ := newProxyLifecycleFixture(t) | |
| 209 | + | transport := newProxyTestTransport() | |
| 210 | + | t.Cleanup(func() { _ = transport.Close() }) | |
| 211 | + | generation, _ := p.installTransport(transport) | |
| 212 | + | ||
| 213 | + | require.False(t, d.markProxyReplaced(p, generation-1, transport)) | |
| 214 | + | require.False(t, proxyExpired(p)) | |
| 215 | + | warm, _ := proxyWarmToken(p) | |
| 216 | + | require.Nil(t, warm, "a rejected replacement must not arm an expiry") | |
| 217 | + | }) | |
| 218 | + | } | |
| 219 | + | ||
| 220 | + | func TestExpireWarmProxyCompletesDaemonOnlyOnLastSession(t *testing.T) { | |
| 221 | + | t.Run("last session completes the daemon", func(t *testing.T) { | |
| 222 | + | d, p, clock := newProxyLifecycleFixture(t) | |
| 223 | + | token, timer := armWarmTimer(t, d, p, clock) | |
| 224 | + | ||
| 225 | + | timer.ch <- time.Time{} | |
| 226 | + | awaitTestCompletion(t, token.done, "warm timer never ran to completion") | |
| 227 | + | awaitTestCompletion(t, p.done, "expiry did not finish the proxy") | |
| 228 | + | awaitTestCompletion(t, d.done, "last session expiry did not complete the daemon") | |
| 229 | + | require.False(t, proxyRegistered(d, p)) | |
| 230 | + | require.True(t, proxyExpired(p)) | |
| 231 | + | }) | |
| 232 | + | ||
| 233 | + | t.Run("surviving session keeps the daemon running", func(t *testing.T) { | |
| 234 | + | d, p, clock := newProxyLifecycleFixture(t) | |
| 235 | + | survivor := registerLifecycleProxy(t, d, domain.RemoteSessionKey{Host: "arch", Name: "other"}) | |
| 236 | + | token, timer := armWarmTimer(t, d, p, clock) | |
| 237 | + | ||
| 238 | + | timer.ch <- time.Time{} | |
| 239 | + | awaitTestCompletion(t, token.done, "warm timer never ran to completion") | |
| 240 | + | require.False(t, proxyRegistered(d, p)) | |
| 241 | + | require.True(t, proxyRegistered(d, survivor)) | |
| 242 | + | select { | |
| 243 | + | case <-d.done: | |
| 244 | + | t.Fatal("daemon completed while a session was still registered") | |
| 245 | + | default: | |
| 246 | + | } | |
| 247 | + | }) | |
| 248 | + | } | |
| 249 | + | ||
| 250 | + | // TestExpireWarmProxyIgnoresSupersededToken proves the expiry callback is exact: | |
| 251 | + | // a token that lost its lifecycle generation must publish no terminal state and | |
| 252 | + | // must not unregister the proxy. | |
| 253 | + | func TestExpireWarmProxyIgnoresSupersededToken(t *testing.T) { | |
| 254 | + | d, p, clock := newProxyLifecycleFixture(t) | |
| 255 | + | stale, _ := armWarmTimer(t, d, p, clock) | |
| 256 | + | current, currentTimer := armWarmTimer(t, d, p, clock) | |
| 257 | + | require.NotSame(t, stale, current, "re-arm did not publish a replacement timer") | |
| 258 | + | awaitTestCompletion(t, stale.done, "superseded warm timer was not canceled") | |
| 259 | + | ||
| 260 | + | require.False(t, d.expireWarmProxy(p, stale), "a superseded token expired the proxy") | |
| 261 | + | require.True(t, proxyRegistered(d, p)) | |
| 262 | + | require.False(t, proxyExpired(p)) | |
| 263 | + | installed, generation := proxyWarmToken(p) | |
| 264 | + | require.Same(t, current, installed) | |
| 265 | + | require.Equal(t, generation, installed.generation) | |
| 266 | + | ||
| 267 | + | currentTimer.ch <- time.Time{} | |
| 268 | + | awaitTestCompletion(t, current.done, "current warm timer never ran to completion") | |
| 269 | + | require.False(t, proxyRegistered(d, p)) | |
| 270 | + | } | |
| @@ -0,0 +1,270 @@ | |||||
| 1 | + | package daemon | |||
| 2 | + | ||||
| 3 | + | import ( | |||
| 4 | + | "sync" | |||
| 5 | + | "testing" | |||
| 6 | + | "time" | |||
| 7 | + | ||||
| 8 | + | "github.com/bnema/vev/internal/domain" | |||
| 9 | + | "github.com/bnema/vev/internal/ports" | |||
| 10 | + | "github.com/stretchr/testify/require" | |||
| 11 | + | ) | |||
| 12 | + | ||||
| 13 | + | // warmHookClock runs a one-shot hook while armProxyWarm sits between its | |||
| 14 | + | // reservation and its publication, so a lifecycle race can be interleaved | |||
| 15 | + | // deterministically instead of being waited for. | |||
| 16 | + | type warmHookClock struct { | |||
| 17 | + | timers chan *signalTimer | |||
| 18 | + | ||||
| 19 | + | mu sync.Mutex | |||
| 20 | + | hook func() | |||
| 21 | + | } | |||
| 22 | + | ||||
| 23 | + | func (c *warmHookClock) Now() time.Time { return time.Time{} } | |||
| 24 | + | ||||
| 25 | + | func (c *warmHookClock) NewTimer(d time.Duration) ports.Timer { | |||
| 26 | + | timer := &signalTimer{ch: make(chan time.Time, 1), duration: d} | |||
| 27 | + | c.mu.Lock() | |||
| 28 | + | hook := c.hook | |||
| 29 | + | c.hook = nil | |||
| 30 | + | c.mu.Unlock() | |||
| 31 | + | if hook != nil { | |||
| 32 | + | hook() | |||
| 33 | + | } | |||
| 34 | + | if c.timers != nil { | |||
| 35 | + | c.timers <- timer | |||
| 36 | + | } | |||
| 37 | + | return timer | |||
| 38 | + | } | |||
| 39 | + | ||||
| 40 | + | func (c *warmHookClock) setHook(hook func()) { | |||
| 41 | + | c.mu.Lock() | |||
| 42 | + | c.hook = hook | |||
| 43 | + | c.mu.Unlock() | |||
| 44 | + | } | |||
| 45 | + | ||||
| 46 | + | func registerLifecycleProxy(t *testing.T, d *Daemon, key domain.RemoteSessionKey) *proxySession { | |||
| 47 | + | t.Helper() | |||
| 48 | + | p, err := newProxySession(key, domain.Size{Cols: 80, Rows: 24}) | |||
| 49 | + | require.NoError(t, err) | |||
| 50 | + | d.mu.Lock() | |||
| 51 | + | registered := d.registerSessionLocked(p) | |||
| 52 | + | d.mu.Unlock() | |||
| 53 | + | require.True(t, registered, "lifecycle fixture proxy was not published") | |||
| 54 | + | return p | |||
| 55 | + | } | |||
| 56 | + | ||||
| 57 | + | func newProxyLifecycleFixture(t *testing.T) (*Daemon, *proxySession, *warmHookClock) { | |||
| 58 | + | t.Helper() | |||
| 59 | + | clock := &warmHookClock{timers: make(chan *signalTimer, 4)} | |||
| 60 | + | d := newProxyTestDaemon(t, nil, clock) | |||
| 61 | + | return d, registerLifecycleProxy(t, d, domain.RemoteSessionKey{Host: "arch", Name: "work"}), clock | |||
| 62 | + | } | |||
| 63 | + | ||||
| 64 | + | func proxyWarmToken(p *proxySession) (*proxyWarmTimer, uint64) { | |||
| 65 | + | p.mu.Lock() | |||
| 66 | + | defer p.mu.Unlock() | |||
| 67 | + | return p.warm, p.generation | |||
| 68 | + | } | |||
| 69 | + | ||||
| 70 | + | func setProxyLifecycleClient(p *proxySession, ac *attachedClient) { | |||
| 71 | + | p.sessionCore.mu.Lock() | |||
| 72 | + | p.client = ac | |||
| 73 | + | p.sessionCore.mu.Unlock() | |||
| 74 | + | } | |||
| 75 | + | ||||
| 76 | + | func proxyRegistered(d *Daemon, p *proxySession) bool { | |||
| 77 | + | d.mu.Lock() | |||
| 78 | + | defer d.mu.Unlock() | |||
| 79 | + | return d.sessions[p.id] == p | |||
| 80 | + | } | |||
| 81 | + | ||||
| 82 | + | func proxyExpired(p *proxySession) bool { | |||
| 83 | + | p.mu.Lock() | |||
| 84 | + | defer p.mu.Unlock() | |||
| 85 | + | return p.expired | |||
| 86 | + | } | |||
| 87 | + | ||||
| 88 | + | func armWarmTimer(t *testing.T, d *Daemon, p *proxySession, clock *warmHookClock) (*proxyWarmTimer, *signalTimer) { | |||
| 89 | + | t.Helper() | |||
| 90 | + | require.True(t, d.armProxyWarm(p), "warm timer was not published") | |||
| 91 | + | timer := awaitTestValue(t, clock.timers, "warm timer was never created") | |||
| 92 | + | token, _ := proxyWarmToken(p) | |||
| 93 | + | require.NotNil(t, token, "published arm left no warm token") | |||
| 94 | + | return token, timer | |||
| 95 | + | } | |||
| 96 | + | ||||
| 97 | + | // TestArmProxyWarmKeepsExpiryPathWhenPublicationFails pins the two-phase arm: | |||
| 98 | + | // a re-arm that loses its revalidation must leave the incumbent timer both | |||
| 99 | + | // installed and current, otherwise a registered proxy is left with no path out | |||
| 100 | + | // of the live registry. | |||
| 101 | + | func TestArmProxyWarmKeepsExpiryPathWhenPublicationFails(t *testing.T) { | |||
| 102 | + | tests := []struct { | |||
| 103 | + | name string | |||
| 104 | + | race func(d *Daemon, p *proxySession) | |||
| 105 | + | undo func(d *Daemon, p *proxySession) | |||
| 106 | + | }{ | |||
| 107 | + | { | |||
| 108 | + | name: "client attaches during arm", | |||
| 109 | + | race: func(_ *Daemon, p *proxySession) { setProxyLifecycleClient(p, &attachedClient{}) }, | |||
| 110 | + | undo: func(_ *Daemon, p *proxySession) { setProxyLifecycleClient(p, nil) }, | |||
| 111 | + | }, | |||
| 112 | + | { | |||
| 113 | + | name: "replacement rekeys the lifecycle generation", | |||
| 114 | + | race: func(_ *Daemon, p *proxySession) { | |||
| 115 | + | p.mu.Lock() | |||
| 116 | + | p.expired = true | |||
| 117 | + | p.generation++ | |||
| 118 | + | if p.warm != nil { | |||
| 119 | + | p.warm.generation = p.generation | |||
| 120 | + | } | |||
| 121 | + | p.mu.Unlock() | |||
| 122 | + | }, | |||
| 123 | + | }, | |||
| 124 | + | { | |||
| 125 | + | name: "daemon starts closing during arm", | |||
| 126 | + | race: func(d *Daemon, _ *proxySession) { | |||
| 127 | + | d.mu.Lock() | |||
| 128 | + | d.closing = true | |||
| 129 | + | d.mu.Unlock() | |||
| 130 | + | }, | |||
| 131 | + | undo: func(d *Daemon, _ *proxySession) { | |||
| 132 | + | d.mu.Lock() | |||
| 133 | + | d.closing = false | |||
| 134 | + | d.mu.Unlock() | |||
| 135 | + | }, | |||
| 136 | + | }, | |||
| 137 | + | } | |||
| 138 | + | for _, tt := range tests { | |||
| 139 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 140 | + | d, p, clock := newProxyLifecycleFixture(t) | |||
| 141 | + | token, timer := armWarmTimer(t, d, p, clock) | |||
| 142 | + | ||||
| 143 | + | clock.setHook(func() { tt.race(d, p) }) | |||
| 144 | + | require.False(t, d.armProxyWarm(p), "raced re-arm must not publish a replacement") | |||
| 145 | + | awaitTestValue(t, clock.timers, "raced re-arm never reached its clock") | |||
| 146 | + | ||||
| 147 | + | retained, generation := proxyWarmToken(p) | |||
| 148 | + | require.NotNil(t, retained, "failed re-arm stripped the proxy of its only expiry path") | |||
| 149 | + | require.Same(t, token, retained, "failed re-arm replaced the armed expiry timer") | |||
| 150 | + | require.Equal(t, generation, retained.generation, "retained timer no longer owns the lifecycle generation") | |||
| 151 | + | require.True(t, proxyRegistered(d, p), "a failed re-arm must not unregister the proxy") | |||
| 152 | + | ||||
| 153 | + | if tt.undo != nil { | |||
| 154 | + | tt.undo(d, p) | |||
| 155 | + | } | |||
| 156 | + | timer.ch <- time.Time{} | |||
| 157 | + | awaitTestCompletion(t, token.done, "retained warm timer never ran to completion") | |||
| 158 | + | require.False(t, proxyRegistered(d, p), "retained warm timer did not expire the dormant proxy") | |||
| 159 | + | require.True(t, proxyExpired(p), "expiry did not publish terminal lifecycle state") | |||
| 160 | + | awaitTestCompletion(t, d.done, "last proxy expiry did not complete the daemon") | |||
| 161 | + | }) | |||
| 162 | + | } | |||
| 163 | + | } | |||
| 164 | + | ||||
| 165 | + | func TestMarkProxyReplacedWarmTimerOwnership(t *testing.T) { | |||
| 166 | + | t.Run("rekeys an already armed timer", func(t *testing.T) { | |||
| 167 | + | d, p, clock := newProxyLifecycleFixture(t) | |||
| 168 | + | transport := newProxyTestTransport() | |||
| 169 | + | t.Cleanup(func() { _ = transport.Close() }) | |||
| 170 | + | generation, _ := p.installTransport(transport) | |||
| 171 | + | token, timer := armWarmTimer(t, d, p, clock) | |||
| 172 | + | ||||
| 173 | + | require.True(t, d.markProxyReplaced(p, generation, transport)) | |||
| 174 | + | require.Empty(t, clock.timers, "replacement armed a second timer instead of rekeying") | |||
| 175 | + | ||||
| 176 | + | rekeyed, lifecycle := proxyWarmToken(p) | |||
| 177 | + | require.Same(t, token, rekeyed, "replacement replaced the armed timer") | |||
| 178 | + | require.Equal(t, lifecycle, rekeyed.generation, "replacement did not rekey the armed timer") | |||
| 179 | + | require.True(t, proxyExpired(p)) | |||
| 180 | + | p.mu.Lock() | |||
| 181 | + | state := p.linkState | |||
| 182 | + | p.mu.Unlock() | |||
| 183 | + | require.Equal(t, ports.LinkStateDead, state) | |||
| 184 | + | ||||
| 185 | + | timer.ch <- time.Time{} | |||
| 186 | + | awaitTestCompletion(t, token.done, "rekeyed warm timer never ran to completion") | |||
| 187 | + | require.False(t, proxyRegistered(d, p), "rekeyed warm timer did not expire the proxy") | |||
| 188 | + | }) | |||
| 189 | + | ||||
| 190 | + | t.Run("arms an expiry for a headless proxy", func(t *testing.T) { | |||
| 191 | + | d, p, clock := newProxyLifecycleFixture(t) | |||
| 192 | + | transport := newProxyTestTransport() | |||
| 193 | + | t.Cleanup(func() { _ = transport.Close() }) | |||
| 194 | + | generation, _ := p.installTransport(transport) | |||
| 195 | + | ||||
| 196 | + | require.True(t, d.markProxyReplaced(p, generation, transport)) | |||
| 197 | + | timer := awaitTestValue(t, clock.timers, "replacement did not arm a warm timer") | |||
| 198 | + | token, lifecycle := proxyWarmToken(p) | |||
| 199 | + | require.NotNil(t, token, "replaced headless proxy has no expiry path") | |||
| 200 | + | require.Equal(t, lifecycle, token.generation) | |||
| 201 | + | ||||
| 202 | + | timer.ch <- time.Time{} | |||
| 203 | + | awaitTestCompletion(t, token.done, "warm timer never ran to completion") | |||
| 204 | + | require.False(t, proxyRegistered(d, p)) | |||
| 205 | + | }) | |||
| 206 | + | ||||
| 207 | + | t.Run("stale link generation publishes nothing", func(t *testing.T) { | |||
| 208 | + | d, p, _ := newProxyLifecycleFixture(t) | |||
| 209 | + | transport := newProxyTestTransport() | |||
| 210 | + | t.Cleanup(func() { _ = transport.Close() }) | |||
| 211 | + | generation, _ := p.installTransport(transport) | |||
| 212 | + | ||||
| 213 | + | require.False(t, d.markProxyReplaced(p, generation-1, transport)) | |||
| 214 | + | require.False(t, proxyExpired(p)) | |||
| 215 | + | warm, _ := proxyWarmToken(p) | |||
| 216 | + | require.Nil(t, warm, "a rejected replacement must not arm an expiry") | |||
| 217 | + | }) | |||
| 218 | + | } | |||
| 219 | + | ||||
| 220 | + | func TestExpireWarmProxyCompletesDaemonOnlyOnLastSession(t *testing.T) { | |||
| 221 | + | t.Run("last session completes the daemon", func(t *testing.T) { | |||
| 222 | + | d, p, clock := newProxyLifecycleFixture(t) | |||
| 223 | + | token, timer := armWarmTimer(t, d, p, clock) | |||
| 224 | + | ||||
| 225 | + | timer.ch <- time.Time{} | |||
| 226 | + | awaitTestCompletion(t, token.done, "warm timer never ran to completion") | |||
| 227 | + | awaitTestCompletion(t, p.done, "expiry did not finish the proxy") | |||
| 228 | + | awaitTestCompletion(t, d.done, "last session expiry did not complete the daemon") | |||
| 229 | + | require.False(t, proxyRegistered(d, p)) | |||
| 230 | + | require.True(t, proxyExpired(p)) | |||
| 231 | + | }) | |||
| 232 | + | ||||
| 233 | + | t.Run("surviving session keeps the daemon running", func(t *testing.T) { | |||
| 234 | + | d, p, clock := newProxyLifecycleFixture(t) | |||
| 235 | + | survivor := registerLifecycleProxy(t, d, domain.RemoteSessionKey{Host: "arch", Name: "other"}) | |||
| 236 | + | token, timer := armWarmTimer(t, d, p, clock) | |||
| 237 | + | ||||
| 238 | + | timer.ch <- time.Time{} | |||
| 239 | + | awaitTestCompletion(t, token.done, "warm timer never ran to completion") | |||
| 240 | + | require.False(t, proxyRegistered(d, p)) | |||
| 241 | + | require.True(t, proxyRegistered(d, survivor)) | |||
| 242 | + | select { | |||
| 243 | + | case <-d.done: | |||
| 244 | + | t.Fatal("daemon completed while a session was still registered") | |||
| 245 | + | default: | |||
| 246 | + | } | |||
| 247 | + | }) | |||
| 248 | + | } | |||
| 249 | + | ||||
| 250 | + | // TestExpireWarmProxyIgnoresSupersededToken proves the expiry callback is exact: | |||
| 251 | + | // a token that lost its lifecycle generation must publish no terminal state and | |||
| 252 | + | // must not unregister the proxy. | |||
| 253 | + | func TestExpireWarmProxyIgnoresSupersededToken(t *testing.T) { | |||
| 254 | + | d, p, clock := newProxyLifecycleFixture(t) | |||
| 255 | + | stale, _ := armWarmTimer(t, d, p, clock) | |||
| 256 | + | current, currentTimer := armWarmTimer(t, d, p, clock) | |||
| 257 | + | require.NotSame(t, stale, current, "re-arm did not publish a replacement timer") | |||
| 258 | + | awaitTestCompletion(t, stale.done, "superseded warm timer was not canceled") | |||
| 259 | + | ||||
| 260 | + | require.False(t, d.expireWarmProxy(p, stale), "a superseded token expired the proxy") | |||
| 261 | + | require.True(t, proxyRegistered(d, p)) | |||
| 262 | + | require.False(t, proxyExpired(p)) | |||
| 263 | + | installed, generation := proxyWarmToken(p) | |||
| 264 | + | require.Same(t, current, installed) | |||
| 265 | + | require.Equal(t, generation, installed.generation) | |||
| 266 | + | ||||
| 267 | + | currentTimer.ch <- time.Time{} | |||
| 268 | + | awaitTestCompletion(t, current.done, "current warm timer never ran to completion") | |||
| 269 | + | require.False(t, proxyRegistered(d, p)) | |||
| 270 | + | } | |||