fix(obs): emit daemon marks outside locks
- commit
- 720c52d83fde960e4c0ef5240995ad80d1ab9e03
- author
- brice · · 2026-07-13 00:31 UTC
- committer
- brice · · 2026-07-13 00:32 UTC
- parent
- 7f91179
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
4 files · +183 -14
M internal/usecase/daemon/render.go +38 -10
| @@ -200,20 +200,44 @@ func (d *Daemon) observeRuntime(kind ports.RuntimeMarkKind, bytes uint64, valid | |||
| 200 | 200 | } | |
| 201 | 201 | } | |
| 202 | 202 | ||
| 203 | - | // runtimeSpan emits endpoints for one real operation. The correlation is | |
| 204 | - | // allocated once before work begins and reused even when the operation fails. | |
| 205 | - | func (d *Daemon) runtimeSpan(start, end ports.RuntimeMarkKind, bytes uint64) func(uint64, bool) { | |
| 206 | - | if d == nil || d.runtimeObserver == nil { | |
| 203 | + | // runtimeMarkBatch captures immutable runtime marks during a render transaction. | |
| 204 | + | // paint owns attachment sendMu while it captures, composes, and emits; JSONL | |
| 205 | + | // observer I/O must therefore happen only after that ownership is released. | |
| 206 | + | type runtimeMarkBatch struct { | |
| 207 | + | observer ports.RuntimeObserver | |
| 208 | + | marks []ports.RuntimeMark | |
| 209 | + | } | |
| 210 | + | ||
| 211 | + | func (d *Daemon) newRuntimeMarkBatch() runtimeMarkBatch { | |
| 212 | + | if d == nil { | |
| 213 | + | return runtimeMarkBatch{} | |
| 214 | + | } | |
| 215 | + | return runtimeMarkBatch{observer: d.runtimeObserver} | |
| 216 | + | } | |
| 217 | + | ||
| 218 | + | func (b *runtimeMarkBatch) span(start, end ports.RuntimeMarkKind, bytes uint64) func(uint64, bool) { | |
| 219 | + | if b == nil || b.observer == nil { | |
| 207 | 220 | return func(uint64, bool) {} | |
| 208 | 221 | } | |
| 209 | 222 | correlation := ports.NewRuntimeCorrelation() | |
| 210 | - | d.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", correlation, start, bytes, true)) | |
| 223 | + | b.marks = append(b.marks, ports.NewRuntimeMarkWithCorrelation("daemon", correlation, start, bytes, true)) | |
| 211 | 224 | return func(endBytes uint64, valid bool) { | |
| 212 | - | d.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", correlation, end, endBytes, valid)) | |
| 225 | + | b.marks = append(b.marks, ports.NewRuntimeMarkWithCorrelation("daemon", correlation, end, endBytes, valid)) | |
| 226 | + | } | |
| 227 | + | } | |
| 228 | + | ||
| 229 | + | func (b *runtimeMarkBatch) flush() { | |
| 230 | + | if b == nil || b.observer == nil { | |
| 231 | + | return | |
| 232 | + | } | |
| 233 | + | for _, mark := range b.marks { | |
| 234 | + | b.observer.ObserveRuntime(mark) | |
| 213 | 235 | } | |
| 214 | 236 | } | |
| 215 | 237 | ||
| 216 | 238 | func (d *Daemon) paint(sess *session, ac *attachedClient, reset bool, epochs ...uint64) { | |
| 239 | + | marks := d.newRuntimeMarkBatch() | |
| 240 | + | ||
| 217 | 241 | attachmentEpoch := uint64(0) | |
| 218 | 242 | if len(epochs) != 0 { | |
| 219 | 243 | attachmentEpoch = epochs[0] | |
| @@ -250,11 +274,15 @@ func (d *Daemon) paint(sess *session, ac *attachedClient, reset bool, epochs ... | |||
| 250 | 274 | overlays := ac.overlays.SnapshotForRender() | |
| 251 | 275 | repaintAttachedClients := false | |
| 252 | 276 | defer func() { | |
| 277 | + | overlays.Unlock() | |
| 278 | + | // Every return below has released sendMu before reaching this defer. | |
| 279 | + | // Emit the captured sequence before any follow-up repaint can introduce | |
| 280 | + | // another boundary, while no attachment, session, tab, or pane lock is held. | |
| 281 | + | marks.flush() | |
| 253 | 282 | if repaintAttachedClients { | |
| 254 | 283 | d.repaintAllAttachedClients() | |
| 255 | 284 | } | |
| 256 | 285 | }() | |
| 257 | - | defer overlays.Unlock() | |
| 258 | 286 | preview := snapshotPickerPreview(nil) | |
| 259 | 287 | if overlays.previewTab != tb { | |
| 260 | 288 | preview = snapshotPickerPreview(overlays.previewTab) | |
| @@ -292,7 +320,7 @@ func (d *Daemon) paint(sess *session, ac *attachedClient, reset bool, epochs ... | |||
| 292 | 320 | pickerActive: overlays.pickerActive, paletteActive: overlays.paletteActive, promptActive: overlays.promptActive, | |
| 293 | 321 | copyFeedback: overlays.copyFeedback, | |
| 294 | 322 | } | |
| 295 | - | endCapture := d.runtimeSpan(ports.RuntimeCaptureStart, ports.RuntimeCaptureEnd, 0) | |
| 323 | + | endCapture := marks.span(ports.RuntimeCaptureStart, ports.RuntimeCaptureEnd, 0) | |
| 296 | 324 | state, ok := captureRenderState(sess, ac, bars, capturedOverlays, preview, floatingCfg, reset, damageCaptureConsume) | |
| 297 | 325 | endCapture(0, ok) | |
| 298 | 326 | if !ok { | |
| @@ -309,13 +337,13 @@ func (d *Daemon) paint(sess *session, ac *attachedClient, reset bool, epochs ... | |||
| 309 | 337 | state.preview = pickerPreviewFromCapturedRender(previewState) | |
| 310 | 338 | } | |
| 311 | 339 | captureOverlayLayers(state, overlays, paletteCfg) | |
| 312 | - | endCompose := d.runtimeSpan(ports.RuntimeComposeStart, ports.RuntimeComposeEnd, 0) | |
| 340 | + | endCompose := marks.span(ports.RuntimeComposeStart, ports.RuntimeComposeEnd, 0) | |
| 313 | 341 | composed := composeFrame(*state, ac.pipelineCache, ac.pipelineScratch) | |
| 314 | 342 | endCompose(0, true) | |
| 315 | 343 | if ac.renderStages.compose != nil { | |
| 316 | 344 | ac.renderStages.compose() | |
| 317 | 345 | } | |
| 318 | - | d.emitFrame(sess, ac, state, composed) | |
| 346 | + | d.emitFrame(sess, ac, state, composed, &marks) | |
| 319 | 347 | } | |
| 320 | 348 | ||
| 321 | 349 | type themeStyles struct { | |
| @@ -200,20 +200,44 @@ func (d *Daemon) observeRuntime(kind ports.RuntimeMarkKind, bytes uint64, valid | |||||
| 200 | } | 200 | } | ||
| 201 | } | 201 | } | ||
| 202 | 202 | ||||
| 203 | - | // runtimeSpan emits endpoints for one real operation. The correlation is | 203 | + | // runtimeMarkBatch captures immutable runtime marks during a render transaction. |
| 204 | - | // allocated once before work begins and reused even when the operation fails. | 204 | + | // paint owns attachment sendMu while it captures, composes, and emits; JSONL |
| 205 | - | func (d *Daemon) runtimeSpan(start, end ports.RuntimeMarkKind, bytes uint64) func(uint64, bool) { | 205 | + | // observer I/O must therefore happen only after that ownership is released. |
| 206 | - | if d == nil || d.runtimeObserver == nil { | 206 | + | type runtimeMarkBatch struct { |
| 207 | + | observer ports.RuntimeObserver | |||
| 208 | + | marks []ports.RuntimeMark | |||
| 209 | + | } | |||
| 210 | + | ||||
| 211 | + | func (d *Daemon) newRuntimeMarkBatch() runtimeMarkBatch { | |||
| 212 | + | if d == nil { | |||
| 213 | + | return runtimeMarkBatch{} | |||
| 214 | + | } | |||
| 215 | + | return runtimeMarkBatch{observer: d.runtimeObserver} | |||
| 216 | + | } | |||
| 217 | + | ||||
| 218 | + | func (b *runtimeMarkBatch) span(start, end ports.RuntimeMarkKind, bytes uint64) func(uint64, bool) { | |||
| 219 | + | if b == nil || b.observer == nil { | |||
| 207 | return func(uint64, bool) {} | 220 | return func(uint64, bool) {} | ||
| 208 | } | 221 | } | ||
| 209 | correlation := ports.NewRuntimeCorrelation() | 222 | correlation := ports.NewRuntimeCorrelation() | ||
| 210 | - | d.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", correlation, start, bytes, true)) | 223 | + | b.marks = append(b.marks, ports.NewRuntimeMarkWithCorrelation("daemon", correlation, start, bytes, true)) |
| 211 | return func(endBytes uint64, valid bool) { | 224 | return func(endBytes uint64, valid bool) { | ||
| 212 | - | d.runtimeObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", correlation, end, endBytes, valid)) | 225 | + | b.marks = append(b.marks, ports.NewRuntimeMarkWithCorrelation("daemon", correlation, end, endBytes, valid)) |
| 226 | + | } | |||
| 227 | + | } | |||
| 228 | + | ||||
| 229 | + | func (b *runtimeMarkBatch) flush() { | |||
| 230 | + | if b == nil || b.observer == nil { | |||
| 231 | + | return | |||
| 232 | + | } | |||
| 233 | + | for _, mark := range b.marks { | |||
| 234 | + | b.observer.ObserveRuntime(mark) | |||
| 213 | } | 235 | } | ||
| 214 | } | 236 | } | ||
| 215 | 237 | ||||
| 216 | func (d *Daemon) paint(sess *session, ac *attachedClient, reset bool, epochs ...uint64) { | 238 | func (d *Daemon) paint(sess *session, ac *attachedClient, reset bool, epochs ...uint64) { | ||
| 239 | + | marks := d.newRuntimeMarkBatch() | |||
| 240 | + | ||||
| 217 | attachmentEpoch := uint64(0) | 241 | attachmentEpoch := uint64(0) | ||
| 218 | if len(epochs) != 0 { | 242 | if len(epochs) != 0 { | ||
| 219 | attachmentEpoch = epochs[0] | 243 | attachmentEpoch = epochs[0] | ||
| @@ -250,11 +274,15 @@ func (d *Daemon) paint(sess *session, ac *attachedClient, reset bool, epochs ... | |||||
| 250 | overlays := ac.overlays.SnapshotForRender() | 274 | overlays := ac.overlays.SnapshotForRender() | ||
| 251 | repaintAttachedClients := false | 275 | repaintAttachedClients := false | ||
| 252 | defer func() { | 276 | defer func() { | ||
| 277 | + | overlays.Unlock() | |||
| 278 | + | // Every return below has released sendMu before reaching this defer. | |||
| 279 | + | // Emit the captured sequence before any follow-up repaint can introduce | |||
| 280 | + | // another boundary, while no attachment, session, tab, or pane lock is held. | |||
| 281 | + | marks.flush() | |||
| 253 | if repaintAttachedClients { | 282 | if repaintAttachedClients { | ||
| 254 | d.repaintAllAttachedClients() | 283 | d.repaintAllAttachedClients() | ||
| 255 | } | 284 | } | ||
| 256 | }() | 285 | }() | ||
| 257 | - | defer overlays.Unlock() | |||
| 258 | preview := snapshotPickerPreview(nil) | 286 | preview := snapshotPickerPreview(nil) | ||
| 259 | if overlays.previewTab != tb { | 287 | if overlays.previewTab != tb { | ||
| 260 | preview = snapshotPickerPreview(overlays.previewTab) | 288 | preview = snapshotPickerPreview(overlays.previewTab) | ||
| @@ -292,7 +320,7 @@ func (d *Daemon) paint(sess *session, ac *attachedClient, reset bool, epochs ... | |||||
| 292 | pickerActive: overlays.pickerActive, paletteActive: overlays.paletteActive, promptActive: overlays.promptActive, | 320 | pickerActive: overlays.pickerActive, paletteActive: overlays.paletteActive, promptActive: overlays.promptActive, | ||
| 293 | copyFeedback: overlays.copyFeedback, | 321 | copyFeedback: overlays.copyFeedback, | ||
| 294 | } | 322 | } | ||
| 295 | - | endCapture := d.runtimeSpan(ports.RuntimeCaptureStart, ports.RuntimeCaptureEnd, 0) | 323 | + | endCapture := marks.span(ports.RuntimeCaptureStart, ports.RuntimeCaptureEnd, 0) |
| 296 | state, ok := captureRenderState(sess, ac, bars, capturedOverlays, preview, floatingCfg, reset, damageCaptureConsume) | 324 | state, ok := captureRenderState(sess, ac, bars, capturedOverlays, preview, floatingCfg, reset, damageCaptureConsume) | ||
| 297 | endCapture(0, ok) | 325 | endCapture(0, ok) | ||
| 298 | if !ok { | 326 | if !ok { | ||
| @@ -309,13 +337,13 @@ func (d *Daemon) paint(sess *session, ac *attachedClient, reset bool, epochs ... | |||||
| 309 | state.preview = pickerPreviewFromCapturedRender(previewState) | 337 | state.preview = pickerPreviewFromCapturedRender(previewState) | ||
| 310 | } | 338 | } | ||
| 311 | captureOverlayLayers(state, overlays, paletteCfg) | 339 | captureOverlayLayers(state, overlays, paletteCfg) | ||
| 312 | - | endCompose := d.runtimeSpan(ports.RuntimeComposeStart, ports.RuntimeComposeEnd, 0) | 340 | + | endCompose := marks.span(ports.RuntimeComposeStart, ports.RuntimeComposeEnd, 0) |
| 313 | composed := composeFrame(*state, ac.pipelineCache, ac.pipelineScratch) | 341 | composed := composeFrame(*state, ac.pipelineCache, ac.pipelineScratch) | ||
| 314 | endCompose(0, true) | 342 | endCompose(0, true) | ||
| 315 | if ac.renderStages.compose != nil { | 343 | if ac.renderStages.compose != nil { | ||
| 316 | ac.renderStages.compose() | 344 | ac.renderStages.compose() | ||
| 317 | } | 345 | } | ||
| 318 | - | d.emitFrame(sess, ac, state, composed) | 346 | + | d.emitFrame(sess, ac, state, composed, &marks) |
| 319 | } | 347 | } | ||
| 320 | 348 | ||||
| 321 | type themeStyles struct { | 349 | type themeStyles struct { | ||
M internal/usecase/daemon/render_coordinator.go +9 -1
| @@ -1096,16 +1096,24 @@ func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) { | |||
| 1096 | 1096 | // notification rather than accumulating ackDeferred forever. | |
| 1097 | 1097 | headlessPreviewOnly := c.detached && c.attachment == nil | |
| 1098 | 1098 | if !headlessPreviewOnly && (!ready || (c.attachment != nil && !c.attachmentReady)) { | |
| 1099 | + | var ackStartObserver ports.RuntimeObserver | |
| 1100 | + | var ackStartCorrelation ports.RuntimeCorrelation | |
| 1099 | 1101 | if !ready && deadline { | |
| 1100 | 1102 | c.ackDeferred = true | |
| 1101 | 1103 | if !c.ackBlocked && c.opts.observer != nil { | |
| 1102 | 1104 | c.ackCorrelation = ports.NewRuntimeCorrelation() | |
| 1103 | 1105 | c.ackBlocked = true | |
| 1104 | - | c.opts.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", c.ackCorrelation, ports.RuntimeACKBlockedStart, 0, true)) | |
| 1106 | + | // Capture the immutable mark while coordinator state is owned, but | |
| 1107 | + | // invoke the potentially blocking observer only after c.mu is free. | |
| 1108 | + | ackStartObserver = c.opts.observer | |
| 1109 | + | ackStartCorrelation = c.ackCorrelation | |
| 1105 | 1110 | } | |
| 1106 | 1111 | } | |
| 1107 | 1112 | preview, previews := c.takePendingPreviewsLocked() | |
| 1108 | 1113 | c.mu.Unlock() | |
| 1114 | + | if ackStartObserver != nil { | |
| 1115 | + | ackStartObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackStartCorrelation, ports.RuntimeACKBlockedStart, 0, true)) | |
| 1116 | + | } | |
| 1109 | 1117 | c.notifyPreviews(w, preview, previews) | |
| 1110 | 1118 | return | |
| 1111 | 1119 | } | |
| @@ -1096,16 +1096,24 @@ func (c *renderCoordinator) fire(gen uint64, watchdog, deadline bool) { | |||||
| 1096 | // notification rather than accumulating ackDeferred forever. | 1096 | // notification rather than accumulating ackDeferred forever. | ||
| 1097 | headlessPreviewOnly := c.detached && c.attachment == nil | 1097 | headlessPreviewOnly := c.detached && c.attachment == nil | ||
| 1098 | if !headlessPreviewOnly && (!ready || (c.attachment != nil && !c.attachmentReady)) { | 1098 | if !headlessPreviewOnly && (!ready || (c.attachment != nil && !c.attachmentReady)) { | ||
| 1099 | + | var ackStartObserver ports.RuntimeObserver | |||
| 1100 | + | var ackStartCorrelation ports.RuntimeCorrelation | |||
| 1099 | if !ready && deadline { | 1101 | if !ready && deadline { | ||
| 1100 | c.ackDeferred = true | 1102 | c.ackDeferred = true | ||
| 1101 | if !c.ackBlocked && c.opts.observer != nil { | 1103 | if !c.ackBlocked && c.opts.observer != nil { | ||
| 1102 | c.ackCorrelation = ports.NewRuntimeCorrelation() | 1104 | c.ackCorrelation = ports.NewRuntimeCorrelation() | ||
| 1103 | c.ackBlocked = true | 1105 | c.ackBlocked = true | ||
| 1104 | - | c.opts.observer.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", c.ackCorrelation, ports.RuntimeACKBlockedStart, 0, true)) | 1106 | + | // Capture the immutable mark while coordinator state is owned, but |
| 1107 | + | // invoke the potentially blocking observer only after c.mu is free. | |||
| 1108 | + | ackStartObserver = c.opts.observer | |||
| 1109 | + | ackStartCorrelation = c.ackCorrelation | |||
| 1105 | } | 1110 | } | ||
| 1106 | } | 1111 | } | ||
| 1107 | preview, previews := c.takePendingPreviewsLocked() | 1112 | preview, previews := c.takePendingPreviewsLocked() | ||
| 1108 | c.mu.Unlock() | 1113 | c.mu.Unlock() | ||
| 1114 | + | if ackStartObserver != nil { | |||
| 1115 | + | ackStartObserver.ObserveRuntime(ports.NewRuntimeMarkWithCorrelation("daemon", ackStartCorrelation, ports.RuntimeACKBlockedStart, 0, true)) | |||
| 1116 | + | } | |||
| 1109 | c.notifyPreviews(w, preview, previews) | 1117 | c.notifyPreviews(w, preview, previews) | ||
| 1110 | return | 1118 | return | ||
| 1111 | } | 1119 | } | ||
M internal/usecase/daemon/render_pipeline.go +14 -3
| @@ -286,7 +286,18 @@ func commitDamageReceipts(receipts []damageReceipt) { | |||
| 286 | 286 | ||
| 287 | 287 | // emitFrame is the sole side-effecting half of the pipeline. The caller holds | |
| 288 | 288 | // sendMu for the complete capture/compose/emit transaction. | |
| 289 | - | func (d *Daemon) emitFrame(sess *session, ac *attachedClient, state *capturedRenderState, composed composedRenderFrame) bool { | |
| 289 | + | func (d *Daemon) emitFrame(sess *session, ac *attachedClient, state *capturedRenderState, composed composedRenderFrame, batches ...*runtimeMarkBatch) bool { | |
| 290 | + | var ownedMarks runtimeMarkBatch | |
| 291 | + | marks := (*runtimeMarkBatch)(nil) | |
| 292 | + | if len(batches) != 0 { | |
| 293 | + | marks = batches[0] | |
| 294 | + | } else { | |
| 295 | + | ownedMarks = d.newRuntimeMarkBatch() | |
| 296 | + | marks = &ownedMarks | |
| 297 | + | // Direct test callers retain the same guarantee as paint: observer I/O | |
| 298 | + | // follows emitFrame's release of attachment ownership. | |
| 299 | + | defer marks.flush() | |
| 300 | + | } | |
| 290 | 301 | if sess == nil || ac == nil || state == nil { | |
| 291 | 302 | if ac != nil { | |
| 292 | 303 | ac.sendMu.Unlock() | |
| @@ -306,7 +317,7 @@ func (d *Daemon) emitFrame(sess *session, ac *attachedClient, state *capturedRen | |||
| 306 | 317 | ac.sendMu.Unlock() | |
| 307 | 318 | return false | |
| 308 | 319 | } | |
| 309 | - | endDiff := d.runtimeSpan(ports.RuntimeDiffStart, ports.RuntimeDiffEnd, 0) | |
| 320 | + | endDiff := marks.span(ports.RuntimeDiffStart, ports.RuntimeDiffEnd, 0) | |
| 310 | 321 | prepared, err := ac.output.prepare(composed.frame, composed.damage, composed.reset) | |
| 311 | 322 | endDiff(0, err == nil) | |
| 312 | 323 | if err != nil { | |
| @@ -321,7 +332,7 @@ func (d *Daemon) emitFrame(sess *session, ac *attachedClient, state *capturedRen | |||
| 321 | 332 | var sendErr error | |
| 322 | 333 | if len(data) > 0 { | |
| 323 | 334 | sendTr = ac.transport() | |
| 324 | - | endEmit := d.runtimeSpan(ports.RuntimeEmitStart, ports.RuntimeEmitEnd, uint64(len(data))) | |
| 335 | + | endEmit := marks.span(ports.RuntimeEmitStart, ports.RuntimeEmitEnd, uint64(len(data))) | |
| 325 | 336 | if sendTr == nil { | |
| 326 | 337 | sendErr = errors.New("client transport is nil") | |
| 327 | 338 | } else { | |
| @@ -286,7 +286,18 @@ func commitDamageReceipts(receipts []damageReceipt) { | |||||
| 286 | 286 | ||||
| 287 | // emitFrame is the sole side-effecting half of the pipeline. The caller holds | 287 | // emitFrame is the sole side-effecting half of the pipeline. The caller holds | ||
| 288 | // sendMu for the complete capture/compose/emit transaction. | 288 | // sendMu for the complete capture/compose/emit transaction. | ||
| 289 | - | func (d *Daemon) emitFrame(sess *session, ac *attachedClient, state *capturedRenderState, composed composedRenderFrame) bool { | 289 | + | func (d *Daemon) emitFrame(sess *session, ac *attachedClient, state *capturedRenderState, composed composedRenderFrame, batches ...*runtimeMarkBatch) bool { |
| 290 | + | var ownedMarks runtimeMarkBatch | |||
| 291 | + | marks := (*runtimeMarkBatch)(nil) | |||
| 292 | + | if len(batches) != 0 { | |||
| 293 | + | marks = batches[0] | |||
| 294 | + | } else { | |||
| 295 | + | ownedMarks = d.newRuntimeMarkBatch() | |||
| 296 | + | marks = &ownedMarks | |||
| 297 | + | // Direct test callers retain the same guarantee as paint: observer I/O | |||
| 298 | + | // follows emitFrame's release of attachment ownership. | |||
| 299 | + | defer marks.flush() | |||
| 300 | + | } | |||
| 290 | if sess == nil || ac == nil || state == nil { | 301 | if sess == nil || ac == nil || state == nil { | ||
| 291 | if ac != nil { | 302 | if ac != nil { | ||
| 292 | ac.sendMu.Unlock() | 303 | ac.sendMu.Unlock() | ||
| @@ -306,7 +317,7 @@ func (d *Daemon) emitFrame(sess *session, ac *attachedClient, state *capturedRen | |||||
| 306 | ac.sendMu.Unlock() | 317 | ac.sendMu.Unlock() | ||
| 307 | return false | 318 | return false | ||
| 308 | } | 319 | } | ||
| 309 | - | endDiff := d.runtimeSpan(ports.RuntimeDiffStart, ports.RuntimeDiffEnd, 0) | 320 | + | endDiff := marks.span(ports.RuntimeDiffStart, ports.RuntimeDiffEnd, 0) |
| 310 | prepared, err := ac.output.prepare(composed.frame, composed.damage, composed.reset) | 321 | prepared, err := ac.output.prepare(composed.frame, composed.damage, composed.reset) | ||
| 311 | endDiff(0, err == nil) | 322 | endDiff(0, err == nil) | ||
| 312 | if err != nil { | 323 | if err != nil { | ||
| @@ -321,7 +332,7 @@ func (d *Daemon) emitFrame(sess *session, ac *attachedClient, state *capturedRen | |||||
| 321 | var sendErr error | 332 | var sendErr error | ||
| 322 | if len(data) > 0 { | 333 | if len(data) > 0 { | ||
| 323 | sendTr = ac.transport() | 334 | sendTr = ac.transport() | ||
| 324 | - | endEmit := d.runtimeSpan(ports.RuntimeEmitStart, ports.RuntimeEmitEnd, uint64(len(data))) | 335 | + | endEmit := marks.span(ports.RuntimeEmitStart, ports.RuntimeEmitEnd, uint64(len(data))) |
| 325 | if sendTr == nil { | 336 | if sendTr == nil { | ||
| 326 | sendErr = errors.New("client transport is nil") | 337 | sendErr = errors.New("client transport is nil") | ||
| 327 | } else { | 338 | } else { | ||
M internal/usecase/daemon/transport_observability_test.go +122 -0
| @@ -1,6 +1,8 @@ | |||
| 1 | 1 | package daemon | |
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | + | "runtime" | |
| 5 | + | "sync" | |
| 4 | 6 | "testing" | |
| 5 | 7 | "time" | |
| 6 | 8 | ||
| @@ -49,3 +51,123 @@ type daemonRuntimeTimer struct{ timer *time.Timer } | |||
| 49 | 51 | func (t daemonRuntimeTimer) C() <-chan time.Time { return t.timer.C } | |
| 50 | 52 | func (t daemonRuntimeTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) } | |
| 51 | 53 | func (t daemonRuntimeTimer) Stop() bool { return t.timer.Stop() } | |
| 54 | + | ||
| 55 | + | type blockingDaemonRuntimeObserver struct { | |
| 56 | + | mu sync.Mutex | |
| 57 | + | entered chan struct{} | |
| 58 | + | release chan struct{} | |
| 59 | + | blocked bool | |
| 60 | + | } | |
| 61 | + | ||
| 62 | + | func newBlockingDaemonRuntimeObserver() *blockingDaemonRuntimeObserver { | |
| 63 | + | return &blockingDaemonRuntimeObserver{entered: make(chan struct{}), release: make(chan struct{})} | |
| 64 | + | } | |
| 65 | + | ||
| 66 | + | func (o *blockingDaemonRuntimeObserver) ObserveRuntime(ports.RuntimeMark) { | |
| 67 | + | o.mu.Lock() | |
| 68 | + | defer o.mu.Unlock() | |
| 69 | + | if !o.blocked { | |
| 70 | + | o.blocked = true | |
| 71 | + | close(o.entered) | |
| 72 | + | <-o.release | |
| 73 | + | } | |
| 74 | + | } | |
| 75 | + | ||
| 76 | + | func awaitDaemonObserver(t *testing.T, ch <-chan struct{}, what string) { | |
| 77 | + | t.Helper() | |
| 78 | + | select { | |
| 79 | + | case <-ch: | |
| 80 | + | case <-time.After(time.Second): | |
| 81 | + | t.Fatalf("timed out waiting for %s", what) | |
| 82 | + | } | |
| 83 | + | } | |
| 84 | + | ||
| 85 | + | // A runtime observer may synchronously write JSONL. Its I/O must run only | |
| 86 | + | // after the render transaction has released attachment, session, and pane | |
| 87 | + | // ownership, so a blocked trace cannot stall live output processing. | |
| 88 | + | func TestTransportObservabilityBlockedRenderObserverReleasesArchitectureLocks(t *testing.T) { | |
| 89 | + | d, sess, ac, sends := newManualSessionWithPTYs(t, nil) | |
| 90 | + | observer := newBlockingDaemonRuntimeObserver() | |
| 91 | + | d.runtimeObserver = observer | |
| 92 | + | ||
| 93 | + | paintDone := make(chan struct{}) | |
| 94 | + | go func() { | |
| 95 | + | d.paint(sess, ac, true) | |
| 96 | + | close(paintDone) | |
| 97 | + | }() | |
| 98 | + | awaitDaemonObserver(t, observer.entered, "blocked render observer") | |
| 99 | + | frame := <-sends // The frame was emitted before the observer began its I/O. | |
| 100 | + | output, err := ports.UnmarshalOutput(frame.Payload) | |
| 101 | + | if err != nil { | |
| 102 | + | t.Fatalf("decode emitted output: %v", err) | |
| 103 | + | } | |
| 104 | + | ac.ackOutputState(output.NewStateNum) | |
| 105 | + | ||
| 106 | + | pane := sess.activeTab().focusedPane() | |
| 107 | + | for _, lock := range []struct { | |
| 108 | + | name string | |
| 109 | + | fn func() | |
| 110 | + | }{ | |
| 111 | + | {"attachment", func() { ac.sendMu.Lock(); runtime.Gosched(); ac.sendMu.Unlock() }}, | |
| 112 | + | {"session", func() { sess.mu.Lock(); runtime.Gosched(); sess.mu.Unlock() }}, | |
| 113 | + | {"pane", func() { pane.mu.Lock(); runtime.Gosched(); pane.mu.Unlock() }}, | |
| 114 | + | } { | |
| 115 | + | done := make(chan struct{}) | |
| 116 | + | go func(fn func()) { fn(); close(done) }(lock.fn) | |
| 117 | + | awaitDaemonObserver(t, done, lock.name+" lock") | |
| 118 | + | } | |
| 119 | + | ||
| 120 | + | livePTY := make(chan struct{}) | |
| 121 | + | go func() { | |
| 122 | + | d.processPTYData(sess, sess.activeTab(), pane, []byte("live"), false) | |
| 123 | + | close(livePTY) | |
| 124 | + | }() | |
| 125 | + | awaitDaemonObserver(t, livePTY, "live PTY processing") | |
| 126 | + | ||
| 127 | + | // A second render can complete its actual pipeline while the first trace | |
| 128 | + | // write is blocked; it is not serialized behind the first attachment lock. | |
| 129 | + | secondPaint := make(chan struct{}) | |
| 130 | + | go func() { | |
| 131 | + | d.paint(sess, ac, false) | |
| 132 | + | close(secondPaint) | |
| 133 | + | }() | |
| 134 | + | <-sends // The second frame is live even though serialized observer I/O waits. | |
| 135 | + | ||
| 136 | + | close(observer.release) | |
| 137 | + | awaitDaemonObserver(t, paintDone, "first render completion") | |
| 138 | + | awaitDaemonObserver(t, secondPaint, "unrelated render completion") | |
| 139 | + | } | |
| 140 | + | ||
| 141 | + | // ACK-blocked start is captured while coordinator state changes, then emitted | |
| 142 | + | // after c.mu is unlocked. This closes the remaining coordinator-observer path. | |
| 143 | + | func TestTransportObservabilityBlockedCoordinatorObserverReleasesLock(t *testing.T) { | |
| 144 | + | observer := newBlockingDaemonRuntimeObserver() | |
| 145 | + | rc := newRenderCoordinator(renderCoordinatorOptions{ | |
| 146 | + | observer: observer, | |
| 147 | + | ackReady: func() bool { return false }, | |
| 148 | + | }) | |
| 149 | + | rc.attach(&attachedClient{}) | |
| 150 | + | rc.mu.Lock() | |
| 151 | + | rc.pending = true | |
| 152 | + | gen := rc.generation | |
| 153 | + | rc.mu.Unlock() | |
| 154 | + | ||
| 155 | + | fired := make(chan struct{}) | |
| 156 | + | go func() { | |
| 157 | + | rc.fire(gen, false, true) | |
| 158 | + | close(fired) | |
| 159 | + | }() | |
| 160 | + | awaitDaemonObserver(t, observer.entered, "blocked ACK observer") | |
| 161 | + | ||
| 162 | + | acquired := make(chan struct{}) | |
| 163 | + | go func() { | |
| 164 | + | rc.mu.Lock() | |
| 165 | + | runtime.Gosched() | |
| 166 | + | rc.mu.Unlock() | |
| 167 | + | close(acquired) | |
| 168 | + | }() | |
| 169 | + | awaitDaemonObserver(t, acquired, "coordinator lock") | |
| 170 | + | ||
| 171 | + | close(observer.release) | |
| 172 | + | awaitDaemonObserver(t, fired, "ACK fire completion") | |
| 173 | + | } | |
| @@ -1,6 +1,8 @@ | |||||
| 1 | package daemon | 1 | package daemon | ||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | + | "runtime" | |||
| 5 | + | "sync" | |||
| 4 | "testing" | 6 | "testing" | ||
| 5 | "time" | 7 | "time" | ||
| 6 | 8 | ||||
| @@ -49,3 +51,123 @@ type daemonRuntimeTimer struct{ timer *time.Timer } | |||||
| 49 | func (t daemonRuntimeTimer) C() <-chan time.Time { return t.timer.C } | 51 | func (t daemonRuntimeTimer) C() <-chan time.Time { return t.timer.C } | ||
| 50 | func (t daemonRuntimeTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) } | 52 | func (t daemonRuntimeTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) } | ||
| 51 | func (t daemonRuntimeTimer) Stop() bool { return t.timer.Stop() } | 53 | func (t daemonRuntimeTimer) Stop() bool { return t.timer.Stop() } | ||
| 54 | + | ||||
| 55 | + | type blockingDaemonRuntimeObserver struct { | |||
| 56 | + | mu sync.Mutex | |||
| 57 | + | entered chan struct{} | |||
| 58 | + | release chan struct{} | |||
| 59 | + | blocked bool | |||
| 60 | + | } | |||
| 61 | + | ||||
| 62 | + | func newBlockingDaemonRuntimeObserver() *blockingDaemonRuntimeObserver { | |||
| 63 | + | return &blockingDaemonRuntimeObserver{entered: make(chan struct{}), release: make(chan struct{})} | |||
| 64 | + | } | |||
| 65 | + | ||||
| 66 | + | func (o *blockingDaemonRuntimeObserver) ObserveRuntime(ports.RuntimeMark) { | |||
| 67 | + | o.mu.Lock() | |||
| 68 | + | defer o.mu.Unlock() | |||
| 69 | + | if !o.blocked { | |||
| 70 | + | o.blocked = true | |||
| 71 | + | close(o.entered) | |||
| 72 | + | <-o.release | |||
| 73 | + | } | |||
| 74 | + | } | |||
| 75 | + | ||||
| 76 | + | func awaitDaemonObserver(t *testing.T, ch <-chan struct{}, what string) { | |||
| 77 | + | t.Helper() | |||
| 78 | + | select { | |||
| 79 | + | case <-ch: | |||
| 80 | + | case <-time.After(time.Second): | |||
| 81 | + | t.Fatalf("timed out waiting for %s", what) | |||
| 82 | + | } | |||
| 83 | + | } | |||
| 84 | + | ||||
| 85 | + | // A runtime observer may synchronously write JSONL. Its I/O must run only | |||
| 86 | + | // after the render transaction has released attachment, session, and pane | |||
| 87 | + | // ownership, so a blocked trace cannot stall live output processing. | |||
| 88 | + | func TestTransportObservabilityBlockedRenderObserverReleasesArchitectureLocks(t *testing.T) { | |||
| 89 | + | d, sess, ac, sends := newManualSessionWithPTYs(t, nil) | |||
| 90 | + | observer := newBlockingDaemonRuntimeObserver() | |||
| 91 | + | d.runtimeObserver = observer | |||
| 92 | + | ||||
| 93 | + | paintDone := make(chan struct{}) | |||
| 94 | + | go func() { | |||
| 95 | + | d.paint(sess, ac, true) | |||
| 96 | + | close(paintDone) | |||
| 97 | + | }() | |||
| 98 | + | awaitDaemonObserver(t, observer.entered, "blocked render observer") | |||
| 99 | + | frame := <-sends // The frame was emitted before the observer began its I/O. | |||
| 100 | + | output, err := ports.UnmarshalOutput(frame.Payload) | |||
| 101 | + | if err != nil { | |||
| 102 | + | t.Fatalf("decode emitted output: %v", err) | |||
| 103 | + | } | |||
| 104 | + | ac.ackOutputState(output.NewStateNum) | |||
| 105 | + | ||||
| 106 | + | pane := sess.activeTab().focusedPane() | |||
| 107 | + | for _, lock := range []struct { | |||
| 108 | + | name string | |||
| 109 | + | fn func() | |||
| 110 | + | }{ | |||
| 111 | + | {"attachment", func() { ac.sendMu.Lock(); runtime.Gosched(); ac.sendMu.Unlock() }}, | |||
| 112 | + | {"session", func() { sess.mu.Lock(); runtime.Gosched(); sess.mu.Unlock() }}, | |||
| 113 | + | {"pane", func() { pane.mu.Lock(); runtime.Gosched(); pane.mu.Unlock() }}, | |||
| 114 | + | } { | |||
| 115 | + | done := make(chan struct{}) | |||
| 116 | + | go func(fn func()) { fn(); close(done) }(lock.fn) | |||
| 117 | + | awaitDaemonObserver(t, done, lock.name+" lock") | |||
| 118 | + | } | |||
| 119 | + | ||||
| 120 | + | livePTY := make(chan struct{}) | |||
| 121 | + | go func() { | |||
| 122 | + | d.processPTYData(sess, sess.activeTab(), pane, []byte("live"), false) | |||
| 123 | + | close(livePTY) | |||
| 124 | + | }() | |||
| 125 | + | awaitDaemonObserver(t, livePTY, "live PTY processing") | |||
| 126 | + | ||||
| 127 | + | // A second render can complete its actual pipeline while the first trace | |||
| 128 | + | // write is blocked; it is not serialized behind the first attachment lock. | |||
| 129 | + | secondPaint := make(chan struct{}) | |||
| 130 | + | go func() { | |||
| 131 | + | d.paint(sess, ac, false) | |||
| 132 | + | close(secondPaint) | |||
| 133 | + | }() | |||
| 134 | + | <-sends // The second frame is live even though serialized observer I/O waits. | |||
| 135 | + | ||||
| 136 | + | close(observer.release) | |||
| 137 | + | awaitDaemonObserver(t, paintDone, "first render completion") | |||
| 138 | + | awaitDaemonObserver(t, secondPaint, "unrelated render completion") | |||
| 139 | + | } | |||
| 140 | + | ||||
| 141 | + | // ACK-blocked start is captured while coordinator state changes, then emitted | |||
| 142 | + | // after c.mu is unlocked. This closes the remaining coordinator-observer path. | |||
| 143 | + | func TestTransportObservabilityBlockedCoordinatorObserverReleasesLock(t *testing.T) { | |||
| 144 | + | observer := newBlockingDaemonRuntimeObserver() | |||
| 145 | + | rc := newRenderCoordinator(renderCoordinatorOptions{ | |||
| 146 | + | observer: observer, | |||
| 147 | + | ackReady: func() bool { return false }, | |||
| 148 | + | }) | |||
| 149 | + | rc.attach(&attachedClient{}) | |||
| 150 | + | rc.mu.Lock() | |||
| 151 | + | rc.pending = true | |||
| 152 | + | gen := rc.generation | |||
| 153 | + | rc.mu.Unlock() | |||
| 154 | + | ||||
| 155 | + | fired := make(chan struct{}) | |||
| 156 | + | go func() { | |||
| 157 | + | rc.fire(gen, false, true) | |||
| 158 | + | close(fired) | |||
| 159 | + | }() | |||
| 160 | + | awaitDaemonObserver(t, observer.entered, "blocked ACK observer") | |||
| 161 | + | ||||
| 162 | + | acquired := make(chan struct{}) | |||
| 163 | + | go func() { | |||
| 164 | + | rc.mu.Lock() | |||
| 165 | + | runtime.Gosched() | |||
| 166 | + | rc.mu.Unlock() | |||
| 167 | + | close(acquired) | |||
| 168 | + | }() | |||
| 169 | + | awaitDaemonObserver(t, acquired, "coordinator lock") | |||
| 170 | + | ||||
| 171 | + | close(observer.release) | |||
| 172 | + | awaitDaemonObserver(t, fired, "ACK fire completion") | |||
| 173 | + | } | |||