fix(daemon): publish sync batches before pane unlock
- commit
- 5747da74f6e5bb810bdca0d571653e63756bbd33
- author
- brice · · 2026-07-11 19:28 UTC
- committer
- brice · · 2026-07-11 19:29 UTC
- parent
- d264a5c
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +83 -52
M internal/usecase/daemon/render.go +22 -13
| @@ -125,6 +125,7 @@ func (d *Daemon) ptyReader(sess *session, tb *tab, p *pane) { | |||
| 125 | 125 | n, err := p.pty.Read(buf) | |
| 126 | 126 | if n > 0 { | |
| 127 | 127 | data := buf[:n] | |
| 128 | + | rc := sess.renderCoordinator() | |
| 128 | 129 | p.mu.Lock() | |
| 129 | 130 | wasSyncing := p.screen.SyncUpdateActive() | |
| 130 | 131 | p.screen.Write(data) | |
| @@ -132,12 +133,28 @@ func (d *Daemon) ptyReader(sess *session, tb *tab, p *pane) { | |||
| 132 | 133 | isSyncing := p.screen.SyncUpdateActive() | |
| 133 | 134 | completeSyncRead := !wasSyncing && !isSyncing && completedSynchronizedUpdate(data) | |
| 134 | 135 | var syncGen uint64 | |
| 136 | + | syncEnded := false | |
| 135 | 137 | if wasSyncing != isSyncing { | |
| 136 | 138 | if isSyncing { | |
| 137 | 139 | syncGen = sess.syncGen.Add(1) | |
| 138 | 140 | p.syncGen = syncGen | |
| 141 | + | // Publish the gate while the authoritative screen mutation is | |
| 142 | + | // still protected. A concurrent deadline therefore cannot | |
| 143 | + | // compose this partial DEC 2026 payload. | |
| 144 | + | if rc != nil { | |
| 145 | + | rc.noteSyncBeginWithRenderability(p, syncGen, func() bool { | |
| 146 | + | return d.paneRenderable(sess, tb, p) | |
| 147 | + | }, func() { | |
| 148 | + | p.mu.Lock() | |
| 149 | + | if p.syncGen == syncGen && p.screen.SyncUpdateActive() { | |
| 150 | + | p.screen.ForceSyncEnd() | |
| 151 | + | } | |
| 152 | + | p.mu.Unlock() | |
| 153 | + | }) | |
| 154 | + | } | |
| 139 | 155 | } else { | |
| 140 | 156 | syncGen = p.syncGen | |
| 157 | + | syncEnded = rc != nil && rc.removeSyncEnd(p, syncGen, true) | |
| 141 | 158 | } | |
| 142 | 159 | } | |
| 143 | 160 | p.mu.Unlock() | |
| @@ -160,26 +177,18 @@ func (d *Daemon) ptyReader(sess *session, tb *tab, p *pane) { | |||
| 160 | 177 | } | |
| 161 | 178 | resp = resp[:0] | |
| 162 | 179 | } | |
| 163 | - | if rc := sess.renderCoordinator(); rc != nil { | |
| 180 | + | if rc != nil { | |
| 164 | 181 | if wasSyncing != isSyncing { | |
| 165 | - | gen := syncGen | |
| 166 | 182 | if isSyncing { | |
| 167 | - | rc.noteSyncBeginWithRenderability(p, gen, func() bool { | |
| 168 | - | return d.paneRenderable(sess, tb, p) | |
| 169 | - | }, func() { | |
| 170 | - | p.mu.Lock() | |
| 171 | - | if p.syncGen == gen && p.screen.SyncUpdateActive() { | |
| 172 | - | p.screen.ForceSyncEnd() | |
| 173 | - | } | |
| 174 | - | p.mu.Unlock() | |
| 175 | - | }) | |
| 176 | 183 | // The accumulated synchronized batch is the pending render only | |
| 177 | 184 | // while this pane can currently reach a composition target. | |
| 178 | 185 | if renderable { | |
| 179 | 186 | rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "render.go"}) | |
| 180 | 187 | } | |
| 181 | - | } else { | |
| 182 | - | rc.noteSyncEnd(p, gen) | |
| 188 | + | } else if syncEnded && renderable { | |
| 189 | + | // State removal happened under pane.mu; only the external wake is | |
| 190 | + | // deferred until after unlock to preserve lock order. | |
| 191 | + | rc.fireCurrent(false) | |
| 183 | 192 | } | |
| 184 | 193 | } | |
| 185 | 194 | if completeSyncRead && renderable { | |
| @@ -125,6 +125,7 @@ func (d *Daemon) ptyReader(sess *session, tb *tab, p *pane) { | |||||
| 125 | n, err := p.pty.Read(buf) | 125 | n, err := p.pty.Read(buf) | ||
| 126 | if n > 0 { | 126 | if n > 0 { | ||
| 127 | data := buf[:n] | 127 | data := buf[:n] | ||
| 128 | + | rc := sess.renderCoordinator() | |||
| 128 | p.mu.Lock() | 129 | p.mu.Lock() | ||
| 129 | wasSyncing := p.screen.SyncUpdateActive() | 130 | wasSyncing := p.screen.SyncUpdateActive() | ||
| 130 | p.screen.Write(data) | 131 | p.screen.Write(data) | ||
| @@ -132,12 +133,28 @@ func (d *Daemon) ptyReader(sess *session, tb *tab, p *pane) { | |||||
| 132 | isSyncing := p.screen.SyncUpdateActive() | 133 | isSyncing := p.screen.SyncUpdateActive() | ||
| 133 | completeSyncRead := !wasSyncing && !isSyncing && completedSynchronizedUpdate(data) | 134 | completeSyncRead := !wasSyncing && !isSyncing && completedSynchronizedUpdate(data) | ||
| 134 | var syncGen uint64 | 135 | var syncGen uint64 | ||
| 136 | + | syncEnded := false | |||
| 135 | if wasSyncing != isSyncing { | 137 | if wasSyncing != isSyncing { | ||
| 136 | if isSyncing { | 138 | if isSyncing { | ||
| 137 | syncGen = sess.syncGen.Add(1) | 139 | syncGen = sess.syncGen.Add(1) | ||
| 138 | p.syncGen = syncGen | 140 | p.syncGen = syncGen | ||
| 141 | + | // Publish the gate while the authoritative screen mutation is | |||
| 142 | + | // still protected. A concurrent deadline therefore cannot | |||
| 143 | + | // compose this partial DEC 2026 payload. | |||
| 144 | + | if rc != nil { | |||
| 145 | + | rc.noteSyncBeginWithRenderability(p, syncGen, func() bool { | |||
| 146 | + | return d.paneRenderable(sess, tb, p) | |||
| 147 | + | }, func() { | |||
| 148 | + | p.mu.Lock() | |||
| 149 | + | if p.syncGen == syncGen && p.screen.SyncUpdateActive() { | |||
| 150 | + | p.screen.ForceSyncEnd() | |||
| 151 | + | } | |||
| 152 | + | p.mu.Unlock() | |||
| 153 | + | }) | |||
| 154 | + | } | |||
| 139 | } else { | 155 | } else { | ||
| 140 | syncGen = p.syncGen | 156 | syncGen = p.syncGen | ||
| 157 | + | syncEnded = rc != nil && rc.removeSyncEnd(p, syncGen, true) | |||
| 141 | } | 158 | } | ||
| 142 | } | 159 | } | ||
| 143 | p.mu.Unlock() | 160 | p.mu.Unlock() | ||
| @@ -160,26 +177,18 @@ func (d *Daemon) ptyReader(sess *session, tb *tab, p *pane) { | |||||
| 160 | } | 177 | } | ||
| 161 | resp = resp[:0] | 178 | resp = resp[:0] | ||
| 162 | } | 179 | } | ||
| 163 | - | if rc := sess.renderCoordinator(); rc != nil { | 180 | + | if rc != nil { |
| 164 | if wasSyncing != isSyncing { | 181 | if wasSyncing != isSyncing { | ||
| 165 | - | gen := syncGen | |||
| 166 | if isSyncing { | 182 | if isSyncing { | ||
| 167 | - | rc.noteSyncBeginWithRenderability(p, gen, func() bool { | |||
| 168 | - | return d.paneRenderable(sess, tb, p) | |||
| 169 | - | }, func() { | |||
| 170 | - | p.mu.Lock() | |||
| 171 | - | if p.syncGen == gen && p.screen.SyncUpdateActive() { | |||
| 172 | - | p.screen.ForceSyncEnd() | |||
| 173 | - | } | |||
| 174 | - | p.mu.Unlock() | |||
| 175 | - | }) | |||
| 176 | // The accumulated synchronized batch is the pending render only | 183 | // The accumulated synchronized batch is the pending render only | ||
| 177 | // while this pane can currently reach a composition target. | 184 | // while this pane can currently reach a composition target. | ||
| 178 | if renderable { | 185 | if renderable { | ||
| 179 | rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "render.go"}) | 186 | rc.invalidate(renderInvalidation{class: invalidateOutput, producer: "render.go"}) | ||
| 180 | } | 187 | } | ||
| 181 | - | } else { | 188 | + | } else if syncEnded && renderable { |
| 182 | - | rc.noteSyncEnd(p, gen) | 189 | + | // State removal happened under pane.mu; only the external wake is |
| 190 | + | // deferred until after unlock to preserve lock order. | |||
| 191 | + | rc.fireCurrent(false) | |||
| 183 | } | 192 | } | ||
| 184 | } | 193 | } | ||
| 185 | if completeSyncRead && renderable { | 194 | if completeSyncRead && renderable { | ||
M internal/usecase/daemon/render_coordinator.go +61 -39
| @@ -84,6 +84,9 @@ type renderCoordinatorOptions struct { | |||
| 84 | 84 | syncActive func() bool | |
| 85 | 85 | // onInvalidate observes every published invalidation (test-visible hook). | |
| 86 | 86 | onInvalidate func(renderInvalidation) | |
| 87 | + | // afterSyncGateEvaluated is a package-private deterministic test seam. It | |
| 88 | + | // runs unlocked after visibility predicates and before registry validation. | |
| 89 | + | afterSyncGateEvaluated func() | |
| 87 | 90 | } | |
| 88 | 91 | ||
| 89 | 92 | // renderCoordinator fans in producer invalidations for one attached session. | |
| @@ -120,6 +123,9 @@ type renderCoordinator struct { | |||
| 120 | 123 | // ackDeferred records that an expired deadline or watchdog was blocked by | |
| 121 | 124 | // output-window capacity. ACK notifications may flush only this state. | |
| 122 | 125 | ackDeferred bool | |
| 126 | + | // deadlineDue closes the handoff between a timer worker receiving its tick | |
| 127 | + | // and fire observing unavailable ACK capacity. | |
| 128 | + | deadlineDue bool | |
| 123 | 129 | coalesced int | |
| 124 | 130 | // outputPressure carries recent bulk-burst pressure between completed | |
| 125 | 131 | // batches. It selects a bounded 8–16ms deadline without letting a busy | |
| @@ -151,6 +157,10 @@ type renderCoordinator struct { | |||
| 151 | 157 | // syncBatches is keyed by the stable pane owner. Each pane owns an | |
| 152 | 158 | // independent watchdog; composition remains gated until every batch ends. | |
| 153 | 159 | syncBatches map[*pane]*syncBatch | |
| 160 | + | // syncRegistryVersion changes for every batch identity mutation. fire | |
| 161 | + | // snapshots it before evaluating visibility and retries if a producer | |
| 162 | + | // publishes or completes a batch during that unlocked evaluation. | |
| 163 | + | syncRegistryVersion uint64 | |
| 154 | 164 | } | |
| 155 | 165 | ||
| 156 | 166 | // newRenderCoordinator constructs a coordinator bound to opts. It starts no | |
| @@ -198,6 +208,7 @@ func (c *renderCoordinator) cancelSyncTimersLocked() { | |||
| 198 | 208 | for pane, batch := range c.syncBatches { | |
| 199 | 209 | cancelSyncBatchLocked(batch) | |
| 200 | 210 | delete(c.syncBatches, pane) | |
| 211 | + | c.syncRegistryVersion++ | |
| 201 | 212 | } | |
| 202 | 213 | } | |
| 203 | 214 | ||
| @@ -215,6 +226,7 @@ func (c *renderCoordinator) invalidate(inv renderInvalidation) { | |||
| 215 | 226 | wasPending, wasUrgent, wasPreviewPending := c.pending, c.pendingUrgent, c.pendingPreview | |
| 216 | 227 | if !wasPending { | |
| 217 | 228 | c.ackDeferred = false | |
| 229 | + | c.deadlineDue = false | |
| 218 | 230 | } | |
| 219 | 231 | c.pending = true | |
| 220 | 232 | c.pendingReset = c.pendingReset || inv.reset | |
| @@ -269,6 +281,7 @@ func (c *renderCoordinator) invalidate(inv renderInvalidation) { | |||
| 269 | 281 | defer close(done) | |
| 270 | 282 | select { | |
| 271 | 283 | case <-timerC: | |
| 284 | + | c.markDeadlineDue(gen) | |
| 272 | 285 | c.fire(gen, false, true) | |
| 273 | 286 | case <-cancel: | |
| 274 | 287 | } | |
| @@ -283,7 +296,7 @@ func (c *renderCoordinator) invalidate(inv renderInvalidation) { | |||
| 283 | 296 | // normal or urgent deadline. | |
| 284 | 297 | func (c *renderCoordinator) notifyAck() { | |
| 285 | 298 | c.mu.Lock() | |
| 286 | - | if !c.ackDeferred || !c.pending { | |
| 299 | + | if (!c.ackDeferred && !c.deadlineDue) || !c.pending { | |
| 287 | 300 | c.mu.Unlock() | |
| 288 | 301 | return | |
| 289 | 302 | } | |
| @@ -294,6 +307,16 @@ func (c *renderCoordinator) notifyAck() { | |||
| 294 | 307 | ||
| 295 | 308 | // noteSyncBegin records a synchronized-output batch for its stable pane and | |
| 296 | 309 | // arms that pane's watchdog. Overlapping pane batches are independent. | |
| 310 | + | // markDeadlineDue records a received timer tick before the worker probes ACK | |
| 311 | + | // readiness, so a concurrent ACK cannot miss an already-expired deadline. | |
| 312 | + | func (c *renderCoordinator) markDeadlineDue(gen uint64) { | |
| 313 | + | c.mu.Lock() | |
| 314 | + | if c.fireValidLocked(gen, false) { | |
| 315 | + | c.deadlineDue = true | |
| 316 | + | } | |
| 317 | + | c.mu.Unlock() | |
| 318 | + | } | |
| 319 | + | ||
| 297 | 320 | func (c *renderCoordinator) noteSyncBegin(p *pane, gen uint64, force ...func()) { | |
| 298 | 321 | var renderable func() bool | |
| 299 | 322 | if c.opts.syncRenderable != nil { | |
| @@ -321,6 +344,7 @@ func (c *renderCoordinator) noteSyncBeginWithRenderability(p *pane, gen uint64, | |||
| 321 | 344 | batch.force = force[0] | |
| 322 | 345 | } | |
| 323 | 346 | c.syncBatches[p] = batch | |
| 347 | + | c.syncRegistryVersion++ | |
| 324 | 348 | clock := c.opts.clock | |
| 325 | 349 | if clock == nil { | |
| 326 | 350 | c.mu.Unlock() | |
| @@ -343,6 +367,7 @@ func (c *renderCoordinator) noteSyncBeginWithRenderability(p *pane, gen uint64, | |||
| 343 | 367 | valid := !c.torndown && current == batch && current.generation == gen | |
| 344 | 368 | if valid { | |
| 345 | 369 | delete(c.syncBatches, p) | |
| 370 | + | c.syncRegistryVersion++ | |
| 346 | 371 | cancelSyncBatchLocked(batch) | |
| 347 | 372 | if len(c.syncBatches) == 0 && c.pending { | |
| 348 | 373 | c.pendingUrgent = true | |
| @@ -362,8 +387,7 @@ func (c *renderCoordinator) noteSyncBeginWithRenderability(p *pane, gen uint64, | |||
| 362 | 387 | // noteSyncEnd records completion for exactly one pane batch. It only flushes | |
| 363 | 388 | // after the aggregate gate opens. | |
| 364 | 389 | func (c *renderCoordinator) noteSyncEnd(p *pane, gen uint64) { | |
| 365 | - | // Visibility may acquire session/tab locks, so sample it before c.mu and | |
| 366 | - | // then verify the batch identity under c.mu. | |
| 390 | + | // Visibility may acquire session/tab locks, so evaluate it outside c.mu. | |
| 367 | 391 | c.mu.Lock() | |
| 368 | 392 | batch := c.syncBatches[p] | |
| 369 | 393 | if batch == nil || batch.generation != gen { | |
| @@ -373,24 +397,29 @@ func (c *renderCoordinator) noteSyncEnd(p *pane, gen uint64) { | |||
| 373 | 397 | renderable := batch.renderable | |
| 374 | 398 | c.mu.Unlock() | |
| 375 | 399 | wasRenderable := renderable == nil || renderable() | |
| 400 | + | if c.removeSyncEnd(p, gen, wasRenderable) { | |
| 401 | + | c.fireCurrent(false) | |
| 402 | + | } | |
| 403 | + | } | |
| 376 | 404 | ||
| 405 | + | // removeSyncEnd removes a batch without invoking a predicate or callback. PTY | |
| 406 | + | // processing calls it while pane.mu is still held, so a deadline can never | |
| 407 | + | // compose the post-Write partial screen between mutation and publication. | |
| 408 | + | // The caller must invoke fireCurrent only after releasing pane.mu. | |
| 409 | + | func (c *renderCoordinator) removeSyncEnd(p *pane, gen uint64, renderable bool) bool { | |
| 377 | 410 | c.mu.Lock() | |
| 378 | - | batch = c.syncBatches[p] | |
| 411 | + | defer c.mu.Unlock() | |
| 412 | + | batch := c.syncBatches[p] | |
| 379 | 413 | if batch == nil || batch.generation != gen { | |
| 380 | - | c.mu.Unlock() | |
| 381 | - | return | |
| 414 | + | return false | |
| 382 | 415 | } | |
| 383 | 416 | delete(c.syncBatches, p) | |
| 417 | + | c.syncRegistryVersion++ | |
| 384 | 418 | cancelSyncBatchLocked(batch) | |
| 385 | - | if wasRenderable && len(c.syncBatches) == 0 && c.pending { | |
| 419 | + | if renderable && len(c.syncBatches) == 0 && c.pending { | |
| 386 | 420 | c.pendingUrgent = true | |
| 387 | 421 | } | |
| 388 | - | c.mu.Unlock() | |
| 389 | - | // A batch that was already non-renderable never closed the aggregate gate; | |
| 390 | - | // its late end must not bypass an unrelated output deadline. | |
| 391 | - | if wasRenderable { | |
| 392 | - | c.fireCurrent(false) | |
| 393 | - | } | |
| 422 | + | return renderable | |
| 394 | 423 | } | |
| 395 | 424 | ||
| 396 | 425 | // noteSyncPaneRemoved releases a pane watchdog when pane lifecycle ends. | |
| @@ -398,6 +427,7 @@ func (c *renderCoordinator) noteSyncPaneRemoved(p *pane) { | |||
| 398 | 427 | c.mu.Lock() | |
| 399 | 428 | if batch := c.syncBatches[p]; batch != nil { | |
| 400 | 429 | delete(c.syncBatches, p) | |
| 430 | + | c.syncRegistryVersion++ | |
| 401 | 431 | cancelSyncBatchLocked(batch) | |
| 402 | 432 | } | |
| 403 | 433 | c.mu.Unlock() | |
| @@ -585,24 +615,20 @@ func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) { | |||
| 585 | 615 | } | |
| 586 | 616 | ackReady := c.opts.ackReady | |
| 587 | 617 | c.mu.Unlock() | |
| 588 | - | if !c.syncGateOpen() { | |
| 589 | - | return | |
| 590 | - | } | |
| 591 | - | ||
| 592 | 618 | ready := ackReady == nil || ackReady() | |
| 593 | 619 | ||
| 594 | - | // The callback can detach, replace, start a synchronized batch, or publish | |
| 595 | - | // newer work. Revalidate every consumption precondition after it returns. | |
| 620 | + | // The readiness callback can detach, replace, or publish a sync batch. | |
| 621 | + | // syncGateOpenLocked retries predicate evaluation on every registry change | |
| 622 | + | // and returns holding c.mu, so pending consumption has no intervening gap. | |
| 596 | 623 | c.mu.Lock() | |
| 597 | 624 | if !c.fireValidLocked(gen, watchdog) { | |
| 598 | 625 | c.mu.Unlock() | |
| 599 | 626 | return | |
| 600 | 627 | } | |
| 601 | - | c.mu.Unlock() | |
| 602 | - | if !c.syncGateOpen() { | |
| 628 | + | if !c.syncGateOpenLocked() { | |
| 629 | + | c.mu.Unlock() | |
| 603 | 630 | return | |
| 604 | 631 | } | |
| 605 | - | c.mu.Lock() | |
| 606 | 632 | if !c.fireValidLocked(gen, watchdog) { | |
| 607 | 633 | c.mu.Unlock() | |
| 608 | 634 | return | |
| @@ -631,7 +657,7 @@ func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) { | |||
| 631 | 657 | } | |
| 632 | 658 | } | |
| 633 | 659 | c.cancelNormalTimerLocked() | |
| 634 | - | c.pending, c.pendingReset, c.pendingUrgent, c.ackDeferred, c.coalesced, c.armed = false, false, false, false, 0, false | |
| 660 | + | c.pending, c.pendingReset, c.pendingUrgent, c.ackDeferred, c.deadlineDue, c.coalesced, c.armed = false, false, false, false, false, 0, false | |
| 635 | 661 | wake := c.opts.wake | |
| 636 | 662 | c.mu.Unlock() | |
| 637 | 663 | if wake != nil { | |
| @@ -646,12 +672,15 @@ func (c *renderCoordinator) fireValidLocked(gen uint64, watchdog bool) bool { | |||
| 646 | 672 | ||
| 647 | 673 | // syncGateOpen evaluates pane visibility without c.mu, then reacquires and | |
| 648 | 674 | // validates the complete batch identity set before accepting that result. | |
| 649 | - | func (c *renderCoordinator) syncGateOpen() bool { | |
| 675 | + | // syncGateOpen evaluates visibility outside c.mu and returns with c.mu held | |
| 676 | + | // only when the registry version still matches the snapshot. The caller must | |
| 677 | + | // consume pending state before unlocking, eliminating the gate-to-consume gap. | |
| 678 | + | func (c *renderCoordinator) syncGateOpenLocked() bool { | |
| 650 | 679 | for { | |
| 651 | - | c.mu.Lock() | |
| 652 | - | batches := make(map[*pane]*syncBatch, len(c.syncBatches)) | |
| 653 | - | for p, batch := range c.syncBatches { | |
| 654 | - | batches[p] = batch | |
| 680 | + | version := c.syncRegistryVersion | |
| 681 | + | batches := make([]*syncBatch, 0, len(c.syncBatches)) | |
| 682 | + | for _, batch := range c.syncBatches { | |
| 683 | + | batches = append(batches, batch) | |
| 655 | 684 | } | |
| 656 | 685 | c.mu.Unlock() | |
| 657 | 686 | ||
| @@ -662,19 +691,12 @@ func (c *renderCoordinator) syncGateOpen() bool { | |||
| 662 | 691 | break | |
| 663 | 692 | } | |
| 664 | 693 | } | |
| 694 | + | if hook := c.opts.afterSyncGateEvaluated; hook != nil { | |
| 695 | + | hook() | |
| 696 | + | } | |
| 665 | 697 | ||
| 666 | 698 | c.mu.Lock() | |
| 667 | - | unchanged := len(batches) == len(c.syncBatches) | |
| 668 | - | if unchanged { | |
| 669 | - | for p, batch := range batches { | |
| 670 | - | if c.syncBatches[p] != batch || c.syncBatches[p].generation != batch.generation { | |
| 671 | - | unchanged = false | |
| 672 | - | break | |
| 673 | - | } | |
| 674 | - | } | |
| 675 | - | } | |
| 676 | - | c.mu.Unlock() | |
| 677 | - | if unchanged { | |
| 699 | + | if version == c.syncRegistryVersion { | |
| 678 | 700 | return !gated | |
| 679 | 701 | } | |
| 680 | 702 | } | |
| @@ -84,6 +84,9 @@ type renderCoordinatorOptions struct { | |||||
| 84 | syncActive func() bool | 84 | syncActive func() bool | ||
| 85 | // onInvalidate observes every published invalidation (test-visible hook). | 85 | // onInvalidate observes every published invalidation (test-visible hook). | ||
| 86 | onInvalidate func(renderInvalidation) | 86 | onInvalidate func(renderInvalidation) | ||
| 87 | + | // afterSyncGateEvaluated is a package-private deterministic test seam. It | |||
| 88 | + | // runs unlocked after visibility predicates and before registry validation. | |||
| 89 | + | afterSyncGateEvaluated func() | |||
| 87 | } | 90 | } | ||
| 88 | 91 | ||||
| 89 | // renderCoordinator fans in producer invalidations for one attached session. | 92 | // renderCoordinator fans in producer invalidations for one attached session. | ||
| @@ -120,6 +123,9 @@ type renderCoordinator struct { | |||||
| 120 | // ackDeferred records that an expired deadline or watchdog was blocked by | 123 | // ackDeferred records that an expired deadline or watchdog was blocked by | ||
| 121 | // output-window capacity. ACK notifications may flush only this state. | 124 | // output-window capacity. ACK notifications may flush only this state. | ||
| 122 | ackDeferred bool | 125 | ackDeferred bool | ||
| 126 | + | // deadlineDue closes the handoff between a timer worker receiving its tick | |||
| 127 | + | // and fire observing unavailable ACK capacity. | |||
| 128 | + | deadlineDue bool | |||
| 123 | coalesced int | 129 | coalesced int | ||
| 124 | // outputPressure carries recent bulk-burst pressure between completed | 130 | // outputPressure carries recent bulk-burst pressure between completed | ||
| 125 | // batches. It selects a bounded 8–16ms deadline without letting a busy | 131 | // batches. It selects a bounded 8–16ms deadline without letting a busy | ||
| @@ -151,6 +157,10 @@ type renderCoordinator struct { | |||||
| 151 | // syncBatches is keyed by the stable pane owner. Each pane owns an | 157 | // syncBatches is keyed by the stable pane owner. Each pane owns an | ||
| 152 | // independent watchdog; composition remains gated until every batch ends. | 158 | // independent watchdog; composition remains gated until every batch ends. | ||
| 153 | syncBatches map[*pane]*syncBatch | 159 | syncBatches map[*pane]*syncBatch | ||
| 160 | + | // syncRegistryVersion changes for every batch identity mutation. fire | |||
| 161 | + | // snapshots it before evaluating visibility and retries if a producer | |||
| 162 | + | // publishes or completes a batch during that unlocked evaluation. | |||
| 163 | + | syncRegistryVersion uint64 | |||
| 154 | } | 164 | } | ||
| 155 | 165 | ||||
| 156 | // newRenderCoordinator constructs a coordinator bound to opts. It starts no | 166 | // newRenderCoordinator constructs a coordinator bound to opts. It starts no | ||
| @@ -198,6 +208,7 @@ func (c *renderCoordinator) cancelSyncTimersLocked() { | |||||
| 198 | for pane, batch := range c.syncBatches { | 208 | for pane, batch := range c.syncBatches { | ||
| 199 | cancelSyncBatchLocked(batch) | 209 | cancelSyncBatchLocked(batch) | ||
| 200 | delete(c.syncBatches, pane) | 210 | delete(c.syncBatches, pane) | ||
| 211 | + | c.syncRegistryVersion++ | |||
| 201 | } | 212 | } | ||
| 202 | } | 213 | } | ||
| 203 | 214 | ||||
| @@ -215,6 +226,7 @@ func (c *renderCoordinator) invalidate(inv renderInvalidation) { | |||||
| 215 | wasPending, wasUrgent, wasPreviewPending := c.pending, c.pendingUrgent, c.pendingPreview | 226 | wasPending, wasUrgent, wasPreviewPending := c.pending, c.pendingUrgent, c.pendingPreview | ||
| 216 | if !wasPending { | 227 | if !wasPending { | ||
| 217 | c.ackDeferred = false | 228 | c.ackDeferred = false | ||
| 229 | + | c.deadlineDue = false | |||
| 218 | } | 230 | } | ||
| 219 | c.pending = true | 231 | c.pending = true | ||
| 220 | c.pendingReset = c.pendingReset || inv.reset | 232 | c.pendingReset = c.pendingReset || inv.reset | ||
| @@ -269,6 +281,7 @@ func (c *renderCoordinator) invalidate(inv renderInvalidation) { | |||||
| 269 | defer close(done) | 281 | defer close(done) | ||
| 270 | select { | 282 | select { | ||
| 271 | case <-timerC: | 283 | case <-timerC: | ||
| 284 | + | c.markDeadlineDue(gen) | |||
| 272 | c.fire(gen, false, true) | 285 | c.fire(gen, false, true) | ||
| 273 | case <-cancel: | 286 | case <-cancel: | ||
| 274 | } | 287 | } | ||
| @@ -283,7 +296,7 @@ func (c *renderCoordinator) invalidate(inv renderInvalidation) { | |||||
| 283 | // normal or urgent deadline. | 296 | // normal or urgent deadline. | ||
| 284 | func (c *renderCoordinator) notifyAck() { | 297 | func (c *renderCoordinator) notifyAck() { | ||
| 285 | c.mu.Lock() | 298 | c.mu.Lock() | ||
| 286 | - | if !c.ackDeferred || !c.pending { | 299 | + | if (!c.ackDeferred && !c.deadlineDue) || !c.pending { |
| 287 | c.mu.Unlock() | 300 | c.mu.Unlock() | ||
| 288 | return | 301 | return | ||
| 289 | } | 302 | } | ||
| @@ -294,6 +307,16 @@ func (c *renderCoordinator) notifyAck() { | |||||
| 294 | 307 | ||||
| 295 | // noteSyncBegin records a synchronized-output batch for its stable pane and | 308 | // noteSyncBegin records a synchronized-output batch for its stable pane and | ||
| 296 | // arms that pane's watchdog. Overlapping pane batches are independent. | 309 | // arms that pane's watchdog. Overlapping pane batches are independent. | ||
| 310 | + | // markDeadlineDue records a received timer tick before the worker probes ACK | |||
| 311 | + | // readiness, so a concurrent ACK cannot miss an already-expired deadline. | |||
| 312 | + | func (c *renderCoordinator) markDeadlineDue(gen uint64) { | |||
| 313 | + | c.mu.Lock() | |||
| 314 | + | if c.fireValidLocked(gen, false) { | |||
| 315 | + | c.deadlineDue = true | |||
| 316 | + | } | |||
| 317 | + | c.mu.Unlock() | |||
| 318 | + | } | |||
| 319 | + | ||||
| 297 | func (c *renderCoordinator) noteSyncBegin(p *pane, gen uint64, force ...func()) { | 320 | func (c *renderCoordinator) noteSyncBegin(p *pane, gen uint64, force ...func()) { | ||
| 298 | var renderable func() bool | 321 | var renderable func() bool | ||
| 299 | if c.opts.syncRenderable != nil { | 322 | if c.opts.syncRenderable != nil { | ||
| @@ -321,6 +344,7 @@ func (c *renderCoordinator) noteSyncBeginWithRenderability(p *pane, gen uint64, | |||||
| 321 | batch.force = force[0] | 344 | batch.force = force[0] | ||
| 322 | } | 345 | } | ||
| 323 | c.syncBatches[p] = batch | 346 | c.syncBatches[p] = batch | ||
| 347 | + | c.syncRegistryVersion++ | |||
| 324 | clock := c.opts.clock | 348 | clock := c.opts.clock | ||
| 325 | if clock == nil { | 349 | if clock == nil { | ||
| 326 | c.mu.Unlock() | 350 | c.mu.Unlock() | ||
| @@ -343,6 +367,7 @@ func (c *renderCoordinator) noteSyncBeginWithRenderability(p *pane, gen uint64, | |||||
| 343 | valid := !c.torndown && current == batch && current.generation == gen | 367 | valid := !c.torndown && current == batch && current.generation == gen | ||
| 344 | if valid { | 368 | if valid { | ||
| 345 | delete(c.syncBatches, p) | 369 | delete(c.syncBatches, p) | ||
| 370 | + | c.syncRegistryVersion++ | |||
| 346 | cancelSyncBatchLocked(batch) | 371 | cancelSyncBatchLocked(batch) | ||
| 347 | if len(c.syncBatches) == 0 && c.pending { | 372 | if len(c.syncBatches) == 0 && c.pending { | ||
| 348 | c.pendingUrgent = true | 373 | c.pendingUrgent = true | ||
| @@ -362,8 +387,7 @@ func (c *renderCoordinator) noteSyncBeginWithRenderability(p *pane, gen uint64, | |||||
| 362 | // noteSyncEnd records completion for exactly one pane batch. It only flushes | 387 | // noteSyncEnd records completion for exactly one pane batch. It only flushes | ||
| 363 | // after the aggregate gate opens. | 388 | // after the aggregate gate opens. | ||
| 364 | func (c *renderCoordinator) noteSyncEnd(p *pane, gen uint64) { | 389 | func (c *renderCoordinator) noteSyncEnd(p *pane, gen uint64) { | ||
| 365 | - | // Visibility may acquire session/tab locks, so sample it before c.mu and | 390 | + | // Visibility may acquire session/tab locks, so evaluate it outside c.mu. |
| 366 | - | // then verify the batch identity under c.mu. | |||
| 367 | c.mu.Lock() | 391 | c.mu.Lock() | ||
| 368 | batch := c.syncBatches[p] | 392 | batch := c.syncBatches[p] | ||
| 369 | if batch == nil || batch.generation != gen { | 393 | if batch == nil || batch.generation != gen { | ||
| @@ -373,24 +397,29 @@ func (c *renderCoordinator) noteSyncEnd(p *pane, gen uint64) { | |||||
| 373 | renderable := batch.renderable | 397 | renderable := batch.renderable | ||
| 374 | c.mu.Unlock() | 398 | c.mu.Unlock() | ||
| 375 | wasRenderable := renderable == nil || renderable() | 399 | wasRenderable := renderable == nil || renderable() | ||
| 400 | + | if c.removeSyncEnd(p, gen, wasRenderable) { | |||
| 401 | + | c.fireCurrent(false) | |||
| 402 | + | } | |||
| 403 | + | } | |||
| 376 | 404 | ||||
| 405 | + | // removeSyncEnd removes a batch without invoking a predicate or callback. PTY | |||
| 406 | + | // processing calls it while pane.mu is still held, so a deadline can never | |||
| 407 | + | // compose the post-Write partial screen between mutation and publication. | |||
| 408 | + | // The caller must invoke fireCurrent only after releasing pane.mu. | |||
| 409 | + | func (c *renderCoordinator) removeSyncEnd(p *pane, gen uint64, renderable bool) bool { | |||
| 377 | c.mu.Lock() | 410 | c.mu.Lock() | ||
| 378 | - | batch = c.syncBatches[p] | 411 | + | defer c.mu.Unlock() |
| 412 | + | batch := c.syncBatches[p] | |||
| 379 | if batch == nil || batch.generation != gen { | 413 | if batch == nil || batch.generation != gen { | ||
| 380 | - | c.mu.Unlock() | 414 | + | return false |
| 381 | - | return | |||
| 382 | } | 415 | } | ||
| 383 | delete(c.syncBatches, p) | 416 | delete(c.syncBatches, p) | ||
| 417 | + | c.syncRegistryVersion++ | |||
| 384 | cancelSyncBatchLocked(batch) | 418 | cancelSyncBatchLocked(batch) | ||
| 385 | - | if wasRenderable && len(c.syncBatches) == 0 && c.pending { | 419 | + | if renderable && len(c.syncBatches) == 0 && c.pending { |
| 386 | c.pendingUrgent = true | 420 | c.pendingUrgent = true | ||
| 387 | } | 421 | } | ||
| 388 | - | c.mu.Unlock() | 422 | + | return renderable |
| 389 | - | // A batch that was already non-renderable never closed the aggregate gate; | |||
| 390 | - | // its late end must not bypass an unrelated output deadline. | |||
| 391 | - | if wasRenderable { | |||
| 392 | - | c.fireCurrent(false) | |||
| 393 | - | } | |||
| 394 | } | 423 | } | ||
| 395 | 424 | ||||
| 396 | // noteSyncPaneRemoved releases a pane watchdog when pane lifecycle ends. | 425 | // noteSyncPaneRemoved releases a pane watchdog when pane lifecycle ends. | ||
| @@ -398,6 +427,7 @@ func (c *renderCoordinator) noteSyncPaneRemoved(p *pane) { | |||||
| 398 | c.mu.Lock() | 427 | c.mu.Lock() | ||
| 399 | if batch := c.syncBatches[p]; batch != nil { | 428 | if batch := c.syncBatches[p]; batch != nil { | ||
| 400 | delete(c.syncBatches, p) | 429 | delete(c.syncBatches, p) | ||
| 430 | + | c.syncRegistryVersion++ | |||
| 401 | cancelSyncBatchLocked(batch) | 431 | cancelSyncBatchLocked(batch) | ||
| 402 | } | 432 | } | ||
| 403 | c.mu.Unlock() | 433 | c.mu.Unlock() | ||
| @@ -585,24 +615,20 @@ func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) { | |||||
| 585 | } | 615 | } | ||
| 586 | ackReady := c.opts.ackReady | 616 | ackReady := c.opts.ackReady | ||
| 587 | c.mu.Unlock() | 617 | c.mu.Unlock() | ||
| 588 | - | if !c.syncGateOpen() { | |||
| 589 | - | return | |||
| 590 | - | } | |||
| 591 | - | ||||
| 592 | ready := ackReady == nil || ackReady() | 618 | ready := ackReady == nil || ackReady() | ||
| 593 | 619 | ||||
| 594 | - | // The callback can detach, replace, start a synchronized batch, or publish | 620 | + | // The readiness callback can detach, replace, or publish a sync batch. |
| 595 | - | // newer work. Revalidate every consumption precondition after it returns. | 621 | + | // syncGateOpenLocked retries predicate evaluation on every registry change |
| 622 | + | // and returns holding c.mu, so pending consumption has no intervening gap. | |||
| 596 | c.mu.Lock() | 623 | c.mu.Lock() | ||
| 597 | if !c.fireValidLocked(gen, watchdog) { | 624 | if !c.fireValidLocked(gen, watchdog) { | ||
| 598 | c.mu.Unlock() | 625 | c.mu.Unlock() | ||
| 599 | return | 626 | return | ||
| 600 | } | 627 | } | ||
| 601 | - | c.mu.Unlock() | 628 | + | if !c.syncGateOpenLocked() { |
| 602 | - | if !c.syncGateOpen() { | 629 | + | c.mu.Unlock() |
| 603 | return | 630 | return | ||
| 604 | } | 631 | } | ||
| 605 | - | c.mu.Lock() | |||
| 606 | if !c.fireValidLocked(gen, watchdog) { | 632 | if !c.fireValidLocked(gen, watchdog) { | ||
| 607 | c.mu.Unlock() | 633 | c.mu.Unlock() | ||
| 608 | return | 634 | return | ||
| @@ -631,7 +657,7 @@ func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) { | |||||
| 631 | } | 657 | } | ||
| 632 | } | 658 | } | ||
| 633 | c.cancelNormalTimerLocked() | 659 | c.cancelNormalTimerLocked() | ||
| 634 | - | c.pending, c.pendingReset, c.pendingUrgent, c.ackDeferred, c.coalesced, c.armed = false, false, false, false, 0, false | 660 | + | c.pending, c.pendingReset, c.pendingUrgent, c.ackDeferred, c.deadlineDue, c.coalesced, c.armed = false, false, false, false, false, 0, false |
| 635 | wake := c.opts.wake | 661 | wake := c.opts.wake | ||
| 636 | c.mu.Unlock() | 662 | c.mu.Unlock() | ||
| 637 | if wake != nil { | 663 | if wake != nil { | ||
| @@ -646,12 +672,15 @@ func (c *renderCoordinator) fireValidLocked(gen uint64, watchdog bool) bool { | |||||
| 646 | 672 | ||||
| 647 | // syncGateOpen evaluates pane visibility without c.mu, then reacquires and | 673 | // syncGateOpen evaluates pane visibility without c.mu, then reacquires and | ||
| 648 | // validates the complete batch identity set before accepting that result. | 674 | // validates the complete batch identity set before accepting that result. | ||
| 649 | - | func (c *renderCoordinator) syncGateOpen() bool { | 675 | + | // syncGateOpen evaluates visibility outside c.mu and returns with c.mu held |
| 676 | + | // only when the registry version still matches the snapshot. The caller must | |||
| 677 | + | // consume pending state before unlocking, eliminating the gate-to-consume gap. | |||
| 678 | + | func (c *renderCoordinator) syncGateOpenLocked() bool { | |||
| 650 | for { | 679 | for { | ||
| 651 | - | c.mu.Lock() | 680 | + | version := c.syncRegistryVersion |
| 652 | - | batches := make(map[*pane]*syncBatch, len(c.syncBatches)) | 681 | + | batches := make([]*syncBatch, 0, len(c.syncBatches)) |
| 653 | - | for p, batch := range c.syncBatches { | 682 | + | for _, batch := range c.syncBatches { |
| 654 | - | batches[p] = batch | 683 | + | batches = append(batches, batch) |
| 655 | } | 684 | } | ||
| 656 | c.mu.Unlock() | 685 | c.mu.Unlock() | ||
| 657 | 686 | ||||
| @@ -662,19 +691,12 @@ func (c *renderCoordinator) syncGateOpen() bool { | |||||
| 662 | break | 691 | break | ||
| 663 | } | 692 | } | ||
| 664 | } | 693 | } | ||
| 694 | + | if hook := c.opts.afterSyncGateEvaluated; hook != nil { | |||
| 695 | + | hook() | |||
| 696 | + | } | |||
| 665 | 697 | ||||
| 666 | c.mu.Lock() | 698 | c.mu.Lock() | ||
| 667 | - | unchanged := len(batches) == len(c.syncBatches) | 699 | + | if version == c.syncRegistryVersion { |
| 668 | - | if unchanged { | |||
| 669 | - | for p, batch := range batches { | |||
| 670 | - | if c.syncBatches[p] != batch || c.syncBatches[p].generation != batch.generation { | |||
| 671 | - | unchanged = false | |||
| 672 | - | break | |||
| 673 | - | } | |||
| 674 | - | } | |||
| 675 | - | } | |||
| 676 | - | c.mu.Unlock() | |||
| 677 | - | if unchanged { | |||
| 678 | return !gated | 700 | return !gated | ||
| 679 | } | 701 | } | ||
| 680 | } | 702 | } | ||