perf(extensions): parallel discovery + auto-ready idle watchdog
Three independent fixes to startup latency: 1) Discover spawns extensions in parallel. Before: each spawn synchronously waited on its child's hello frame; multiple slow runtimes (e.g. tsx) added linearly. After: every loadOne runs on its own goroutine; total time collapses to max(spawn_time) instead of sum. 2) WaitForReady waits in parallel. Before: one extension at a time, so a slow ready (or no ready at all from a legacy SDK) blocked every other extension's wait too. After: one goroutine per extension, all sharing a single deadline; total = max(per-ext wait), not sum. 3) Auto-ready idle watchdog for legacy extensions. Phase-1 SDK builds didn't send the ready sentinel introduced in phase 2. Without it, WaitForReady burned the full 3s grace on every startup for every legacy extension. Fix: read loop stamps lastFrameTime on each frame; a per-extension watchdog closes readyCh as soon as no new frame has arrived for 250ms. Native binaries register + go quiet within microseconds, so this fires almost immediately. Newer extensions still trip the explicit ready path before the watchdog matters. Also updates the scratchpad example to invoke `tsx` directly instead of `npx --yes tsx`, with the README explaining how to install tsx globally and how to fall back to npx (and what it costs in startup time). Measured impact on a machine with 4 extensions installed (guard / hello / weather / scratchpad): before: 4.2-4.9s per zot launch after: ~200ms per zot launch (cold-cache first run ~780ms) The dominant remaining cost in the 200ms is normal node + tsx boot for scratchpad, which only matters because it's still in the spawn fan-out — Go extensions add nothing measurable.
- commit
- 619ed587cdea62b9d8a57ab9f8942c414500ced8
- author
- patriceckhart · · 2026-04-19 13:13 UTC
- parent
- 5dbbcb9
- signature
- unsigned
3 files · +108 -20
M examples/extensions/scratchpad/README.md +5 -5
| @@ -13,16 +13,16 @@ Demonstrates: | |||
| 13 | 13 | ||
| 14 | 14 | ## Requirements | |
| 15 | 15 | ||
| 16 | - | Node 18+. The first time you use it, `npx` will quietly fetch `tsx` | |
| 17 | - | into the npm cache (~5 MB, one-shot). If you'd rather pre-install | |
| 18 | - | globally: | |
| 16 | + | Node 18+ and `tsx` on `$PATH`: | |
| 19 | 17 | ||
| 20 | 18 | ```bash | |
| 21 | 19 | npm install -g tsx | |
| 22 | 20 | ``` | |
| 23 | 21 | ||
| 24 | - | …and change `extension.json`'s `exec` to `"tsx"` and `args` to | |
| 25 | - | `["index.ts"]` if you want. | |
| 22 | + | This is what `extension.json` invokes (`exec: "tsx"`). Without the | |
| 23 | + | global install, swap to `"exec": "npx"` + `"args": ["--yes", "tsx", | |
| 24 | + | "index.ts"]` — functional but adds ~1 second to each zot startup | |
| 25 | + | because npm checks the registry on every invocation. | |
| 26 | 26 | ||
| 27 | 27 | ## Install | |
| 28 | 28 | ||
| @@ -13,16 +13,16 @@ Demonstrates: | |||||
| 13 | 13 | ||||
| 14 | ## Requirements | 14 | ## Requirements | ||
| 15 | 15 | ||||
| 16 | - | Node 18+. The first time you use it, `npx` will quietly fetch `tsx` | 16 | + | Node 18+ and `tsx` on `$PATH`: |
| 17 | - | into the npm cache (~5 MB, one-shot). If you'd rather pre-install | |||
| 18 | - | globally: | |||
| 19 | 17 | ||||
| 20 | ```bash | 18 | ```bash | ||
| 21 | npm install -g tsx | 19 | npm install -g tsx | ||
| 22 | ``` | 20 | ``` | ||
| 23 | 21 | ||||
| 24 | - | …and change `extension.json`'s `exec` to `"tsx"` and `args` to | 22 | + | This is what `extension.json` invokes (`exec: "tsx"`). Without the |
| 25 | - | `["index.ts"]` if you want. | 23 | + | global install, swap to `"exec": "npx"` + `"args": ["--yes", "tsx", |
| 24 | + | "index.ts"]` — functional but adds ~1 second to each zot startup | |||
| 25 | + | because npm checks the registry on every invocation. | |||
| 26 | 26 | ||||
| 27 | ## Install | 27 | ## Install | ||
| 28 | 28 | ||||
M examples/extensions/scratchpad/extension.json +2 -2
| @@ -1,8 +1,8 @@ | |||
| 1 | 1 | { | |
| 2 | 2 | "name": "scratchpad", | |
| 3 | 3 | "version": "1.0.0", | |
| 4 | - | "exec": "npx", | |
| 5 | - | "args": ["--yes", "tsx", "index.ts"], | |
| 4 | + | "exec": "tsx", | |
| 5 | + | "args": ["index.ts"], | |
| 6 | 6 | "language": "typescript", | |
| 7 | 7 | "description": "scratchpad notes (commands) + read_notes (tool) — written in plain .ts via tsx", | |
| 8 | 8 | "enabled": true | |
| @@ -1,8 +1,8 @@ | |||||
| 1 | { | 1 | { | ||
| 2 | "name": "scratchpad", | 2 | "name": "scratchpad", | ||
| 3 | "version": "1.0.0", | 3 | "version": "1.0.0", | ||
| 4 | - | "exec": "npx", | 4 | + | "exec": "tsx", |
| 5 | - | "args": ["--yes", "tsx", "index.ts"], | 5 | + | "args": ["index.ts"], |
| 6 | "language": "typescript", | 6 | "language": "typescript", | ||
| 7 | "description": "scratchpad notes (commands) + read_notes (tool) — written in plain .ts via tsx", | 7 | "description": "scratchpad notes (commands) + read_notes (tool) — written in plain .ts via tsx", | ||
| 8 | "enabled": true | 8 | "enabled": true | ||
M internal/agent/extensions/manager.go +101 -13
| @@ -79,6 +79,12 @@ type Extension struct { | |||
| 79 | 79 | pendingTool map[string]chan extproto.ToolResultFromExt | |
| 80 | 80 | pendingIntercept map[string]chan extproto.EventInterceptResponseFromExt | |
| 81 | 81 | ||
| 82 | + | // lastFrameTime is updated by the read loop on every frame it | |
| 83 | + | // processes. Used by the auto-ready idle watchdog so legacy | |
| 84 | + | // extensions (no `ready` frame) don't pin the WaitForReady wait | |
| 85 | + | // to its full grace. | |
| 86 | + | lastFrameTime time.Time | |
| 87 | + | ||
| 82 | 88 | // eventSubs and interceptSubs are the sets of event names this | |
| 83 | 89 | // extension subscribed to via SubscribeFromExt. Used by | |
| 84 | 90 | // EmitEvent / InterceptToolCall to filter recipients. | |
| @@ -145,11 +151,17 @@ func New(zotHome, cwd, zotVersion, provider, model string, hooks HostHooks) *Man | |||
| 145 | 151 | } | |
| 146 | 152 | ||
| 147 | 153 | // Discover scans the global and project extension dirs and starts | |
| 148 | - | // every extension whose manifest is enabled. Returns a slice of | |
| 154 | + | // every extension whose manifest is enabled. Spawns happen in | |
| 155 | + | // parallel so a slow runtime (e.g. `npx tsx` cold-start, ~1.5s) | |
| 156 | + | // doesn't block other extensions from starting. Returns a slice of | |
| 149 | 157 | // errors encountered (one per extension); a single bad extension | |
| 150 | 158 | // doesn't abort the rest. | |
| 151 | 159 | func (m *Manager) Discover(ctx context.Context) []error { | |
| 152 | - | var errs []error | |
| 160 | + | type loadJob struct { | |
| 161 | + | dir string | |
| 162 | + | } | |
| 163 | + | var jobs []loadJob | |
| 164 | + | seenDirs := map[string]bool{} // dedup by basename so project wins | |
| 153 | 165 | for _, dir := range m.searchDirs() { | |
| 154 | 166 | entries, err := os.ReadDir(dir) | |
| 155 | 167 | if err != nil { | |
| @@ -159,12 +171,32 @@ func (m *Manager) Discover(ctx context.Context) []error { | |||
| 159 | 171 | if !e.IsDir() { | |
| 160 | 172 | continue | |
| 161 | 173 | } | |
| 162 | - | extDir := filepath.Join(dir, e.Name()) | |
| 163 | - | if err := m.loadOne(ctx, extDir); err != nil { | |
| 164 | - | errs = append(errs, fmt.Errorf("%s: %w", extDir, err)) | |
| 174 | + | if seenDirs[e.Name()] { | |
| 175 | + | continue // higher-priority location already queued | |
| 165 | 176 | } | |
| 177 | + | seenDirs[e.Name()] = true | |
| 178 | + | jobs = append(jobs, loadJob{dir: filepath.Join(dir, e.Name())}) | |
| 166 | 179 | } | |
| 167 | 180 | } | |
| 181 | + | ||
| 182 | + | var wg sync.WaitGroup | |
| 183 | + | errCh := make(chan error, len(jobs)) | |
| 184 | + | for _, j := range jobs { | |
| 185 | + | wg.Add(1) | |
| 186 | + | go func(extDir string) { | |
| 187 | + | defer wg.Done() | |
| 188 | + | if err := m.loadOne(ctx, extDir); err != nil { | |
| 189 | + | errCh <- fmt.Errorf("%s: %w", extDir, err) | |
| 190 | + | } | |
| 191 | + | }(j.dir) | |
| 192 | + | } | |
| 193 | + | wg.Wait() | |
| 194 | + | close(errCh) | |
| 195 | + | ||
| 196 | + | var errs []error | |
| 197 | + | for e := range errCh { | |
| 198 | + | errs = append(errs, e) | |
| 199 | + | } | |
| 168 | 200 | return errs | |
| 169 | 201 | } | |
| 170 | 202 | ||
| @@ -237,8 +269,14 @@ func (m *Manager) loadOne(ctx context.Context, dir string) error { | |||
| 237 | 269 | return nil | |
| 238 | 270 | } | |
| 239 | 271 | ||
| 240 | - | // WaitForReady blocks until every loaded extension has sent its | |
| 241 | - | // ReadyFromExt sentinel, or the per-extension grace period expires. | |
| 272 | + | // WaitForReady blocks until every loaded extension has signalled | |
| 273 | + | // ReadyFromExt, or the grace period expires for the slowest one. | |
| 274 | + | // | |
| 275 | + | // Waits run in parallel: total time is max(per-extension wait), not | |
| 276 | + | // sum. Without this, a single slow extension (e.g. `npx tsx` cold) | |
| 277 | + | // would gate every other extension's wait too and zot startup would | |
| 278 | + | // scale linearly with the number of slow runtimes installed. | |
| 279 | + | // | |
| 242 | 280 | // Call after Discover and before relying on tool registrations. | |
| 243 | 281 | func (m *Manager) WaitForReady(grace time.Duration) { | |
| 244 | 282 | m.mu.RLock() | |
| @@ -247,14 +285,22 @@ func (m *Manager) WaitForReady(grace time.Duration) { | |||
| 247 | 285 | exts = append(exts, e) | |
| 248 | 286 | } | |
| 249 | 287 | m.mu.RUnlock() | |
| 288 | + | ||
| 289 | + | deadline := time.After(grace) | |
| 290 | + | var wg sync.WaitGroup | |
| 250 | 291 | for _, ext := range exts { | |
| 251 | - | select { | |
| 252 | - | case <-ext.readyCh: | |
| 253 | - | case <-time.After(grace): | |
| 254 | - | fmt.Fprintf(ext.logFile, "[zot] timed out waiting for ready frame; proceeding\n") | |
| 255 | - | ext.readyOnce.Do(func() { close(ext.readyCh) }) | |
| 256 | - | } | |
| 292 | + | wg.Add(1) | |
| 293 | + | go func(ext *Extension) { | |
| 294 | + | defer wg.Done() | |
| 295 | + | select { | |
| 296 | + | case <-ext.readyCh: | |
| 297 | + | case <-deadline: | |
| 298 | + | fmt.Fprintf(ext.logFile, "[zot] timed out waiting for ready frame; proceeding\n") | |
| 299 | + | ext.readyOnce.Do(func() { close(ext.readyCh) }) | |
| 300 | + | } | |
| 301 | + | }(ext) | |
| 257 | 302 | } | |
| 303 | + | wg.Wait() | |
| 258 | 304 | } | |
| 259 | 305 | ||
| 260 | 306 | // spawn launches the subprocess, hooks up pipes, logs stderr, and | |
| @@ -344,9 +390,48 @@ func (m *Manager) spawn(ctx context.Context, ext *Extension) error { | |||
| 344 | 390 | ||
| 345 | 391 | // Spin up the read loop now that the handshake is done. | |
| 346 | 392 | go m.readLoop(ext, scanner) | |
| 393 | + | ||
| 394 | + | // Compatibility shim: extensions built against the phase-1 SDK | |
| 395 | + | // don't send a ready frame. Watch the read loop's frame arrival | |
| 396 | + | // rate; if nothing's arrived for readyIdleWindow we treat the | |
| 397 | + | // extension as ready so WaitForReady doesn't burn the full grace | |
| 398 | + | // on every startup. Newer extensions still trigger the explicit | |
| 399 | + | // path on their own ready frame. | |
| 400 | + | go m.assumeReadyAfterIdle(ext) | |
| 401 | + | ||
| 347 | 402 | return nil | |
| 348 | 403 | } | |
| 349 | 404 | ||
| 405 | + | // readyIdleWindow is how long the manager waits for a frame after | |
| 406 | + | // hello before assuming an extension that doesn't send `ready` is | |
| 407 | + | // nevertheless ready. 250ms is enough for any well-behaved native | |
| 408 | + | // binary to flush its register frames; slow runtimes (npx tsx) flush | |
| 409 | + | // even faster once they've started, so this rarely affects them. | |
| 410 | + | const readyIdleWindow = 250 * time.Millisecond | |
| 411 | + | ||
| 412 | + | func (m *Manager) assumeReadyAfterIdle(ext *Extension) { | |
| 413 | + | last := ext.lastFrameTime | |
| 414 | + | for { | |
| 415 | + | select { | |
| 416 | + | case <-ext.readyCh: | |
| 417 | + | return | |
| 418 | + | case <-time.After(readyIdleWindow): | |
| 419 | + | } | |
| 420 | + | ext.mu.Lock() | |
| 421 | + | current := ext.lastFrameTime | |
| 422 | + | ext.mu.Unlock() | |
| 423 | + | if current.Equal(last) { | |
| 424 | + | // No new frame in the idle window. Treat as ready. | |
| 425 | + | ext.readyOnce.Do(func() { | |
| 426 | + | fmt.Fprintf(ext.logFile, "[zot] no ready frame; auto-readying after idle (legacy SDK?)\n") | |
| 427 | + | close(ext.readyCh) | |
| 428 | + | }) | |
| 429 | + | return | |
| 430 | + | } | |
| 431 | + | last = current | |
| 432 | + | } | |
| 433 | + | } | |
| 434 | + | ||
| 350 | 435 | // readLoop processes every frame the extension sends after hello. | |
| 351 | 436 | // Returns when stdout closes. | |
| 352 | 437 | func (m *Manager) readLoop(ext *Extension, scanner *bufio.Scanner) { | |
| @@ -372,6 +457,9 @@ func (m *Manager) readLoop(ext *Extension, scanner *bufio.Scanner) { | |||
| 372 | 457 | ||
| 373 | 458 | for scanner.Scan() { | |
| 374 | 459 | line := scanner.Bytes() | |
| 460 | + | ext.mu.Lock() | |
| 461 | + | ext.lastFrameTime = time.Now() | |
| 462 | + | ext.mu.Unlock() | |
| 375 | 463 | var frame extproto.Frame | |
| 376 | 464 | if err := json.Unmarshal(line, &frame); err != nil { | |
| 377 | 465 | fmt.Fprintf(ext.logFile, "[zot] malformed json from extension: %v\n", err) | |
| @@ -79,6 +79,12 @@ type Extension struct { | |||||
| 79 | pendingTool map[string]chan extproto.ToolResultFromExt | 79 | pendingTool map[string]chan extproto.ToolResultFromExt | ||
| 80 | pendingIntercept map[string]chan extproto.EventInterceptResponseFromExt | 80 | pendingIntercept map[string]chan extproto.EventInterceptResponseFromExt | ||
| 81 | 81 | ||||
| 82 | + | // lastFrameTime is updated by the read loop on every frame it | |||
| 83 | + | // processes. Used by the auto-ready idle watchdog so legacy | |||
| 84 | + | // extensions (no `ready` frame) don't pin the WaitForReady wait | |||
| 85 | + | // to its full grace. | |||
| 86 | + | lastFrameTime time.Time | |||
| 87 | + | ||||
| 82 | // eventSubs and interceptSubs are the sets of event names this | 88 | // eventSubs and interceptSubs are the sets of event names this | ||
| 83 | // extension subscribed to via SubscribeFromExt. Used by | 89 | // extension subscribed to via SubscribeFromExt. Used by | ||
| 84 | // EmitEvent / InterceptToolCall to filter recipients. | 90 | // EmitEvent / InterceptToolCall to filter recipients. | ||
| @@ -145,11 +151,17 @@ func New(zotHome, cwd, zotVersion, provider, model string, hooks HostHooks) *Man | |||||
| 145 | } | 151 | } | ||
| 146 | 152 | ||||
| 147 | // Discover scans the global and project extension dirs and starts | 153 | // Discover scans the global and project extension dirs and starts | ||
| 148 | - | // every extension whose manifest is enabled. Returns a slice of | 154 | + | // every extension whose manifest is enabled. Spawns happen in |
| 155 | + | // parallel so a slow runtime (e.g. `npx tsx` cold-start, ~1.5s) | |||
| 156 | + | // doesn't block other extensions from starting. Returns a slice of | |||
| 149 | // errors encountered (one per extension); a single bad extension | 157 | // errors encountered (one per extension); a single bad extension | ||
| 150 | // doesn't abort the rest. | 158 | // doesn't abort the rest. | ||
| 151 | func (m *Manager) Discover(ctx context.Context) []error { | 159 | func (m *Manager) Discover(ctx context.Context) []error { | ||
| 152 | - | var errs []error | 160 | + | type loadJob struct { |
| 161 | + | dir string | |||
| 162 | + | } | |||
| 163 | + | var jobs []loadJob | |||
| 164 | + | seenDirs := map[string]bool{} // dedup by basename so project wins | |||
| 153 | for _, dir := range m.searchDirs() { | 165 | for _, dir := range m.searchDirs() { | ||
| 154 | entries, err := os.ReadDir(dir) | 166 | entries, err := os.ReadDir(dir) | ||
| 155 | if err != nil { | 167 | if err != nil { | ||
| @@ -159,12 +171,32 @@ func (m *Manager) Discover(ctx context.Context) []error { | |||||
| 159 | if !e.IsDir() { | 171 | if !e.IsDir() { | ||
| 160 | continue | 172 | continue | ||
| 161 | } | 173 | } | ||
| 162 | - | extDir := filepath.Join(dir, e.Name()) | 174 | + | if seenDirs[e.Name()] { |
| 163 | - | if err := m.loadOne(ctx, extDir); err != nil { | 175 | + | continue // higher-priority location already queued |
| 164 | - | errs = append(errs, fmt.Errorf("%s: %w", extDir, err)) | |||
| 165 | } | 176 | } | ||
| 177 | + | seenDirs[e.Name()] = true | |||
| 178 | + | jobs = append(jobs, loadJob{dir: filepath.Join(dir, e.Name())}) | |||
| 166 | } | 179 | } | ||
| 167 | } | 180 | } | ||
| 181 | + | ||||
| 182 | + | var wg sync.WaitGroup | |||
| 183 | + | errCh := make(chan error, len(jobs)) | |||
| 184 | + | for _, j := range jobs { | |||
| 185 | + | wg.Add(1) | |||
| 186 | + | go func(extDir string) { | |||
| 187 | + | defer wg.Done() | |||
| 188 | + | if err := m.loadOne(ctx, extDir); err != nil { | |||
| 189 | + | errCh <- fmt.Errorf("%s: %w", extDir, err) | |||
| 190 | + | } | |||
| 191 | + | }(j.dir) | |||
| 192 | + | } | |||
| 193 | + | wg.Wait() | |||
| 194 | + | close(errCh) | |||
| 195 | + | ||||
| 196 | + | var errs []error | |||
| 197 | + | for e := range errCh { | |||
| 198 | + | errs = append(errs, e) | |||
| 199 | + | } | |||
| 168 | return errs | 200 | return errs | ||
| 169 | } | 201 | } | ||
| 170 | 202 | ||||
| @@ -237,8 +269,14 @@ func (m *Manager) loadOne(ctx context.Context, dir string) error { | |||||
| 237 | return nil | 269 | return nil | ||
| 238 | } | 270 | } | ||
| 239 | 271 | ||||
| 240 | - | // WaitForReady blocks until every loaded extension has sent its | 272 | + | // WaitForReady blocks until every loaded extension has signalled |
| 241 | - | // ReadyFromExt sentinel, or the per-extension grace period expires. | 273 | + | // ReadyFromExt, or the grace period expires for the slowest one. |
| 274 | + | // | |||
| 275 | + | // Waits run in parallel: total time is max(per-extension wait), not | |||
| 276 | + | // sum. Without this, a single slow extension (e.g. `npx tsx` cold) | |||
| 277 | + | // would gate every other extension's wait too and zot startup would | |||
| 278 | + | // scale linearly with the number of slow runtimes installed. | |||
| 279 | + | // | |||
| 242 | // Call after Discover and before relying on tool registrations. | 280 | // Call after Discover and before relying on tool registrations. | ||
| 243 | func (m *Manager) WaitForReady(grace time.Duration) { | 281 | func (m *Manager) WaitForReady(grace time.Duration) { | ||
| 244 | m.mu.RLock() | 282 | m.mu.RLock() | ||
| @@ -247,14 +285,22 @@ func (m *Manager) WaitForReady(grace time.Duration) { | |||||
| 247 | exts = append(exts, e) | 285 | exts = append(exts, e) | ||
| 248 | } | 286 | } | ||
| 249 | m.mu.RUnlock() | 287 | m.mu.RUnlock() | ||
| 288 | + | ||||
| 289 | + | deadline := time.After(grace) | |||
| 290 | + | var wg sync.WaitGroup | |||
| 250 | for _, ext := range exts { | 291 | for _, ext := range exts { | ||
| 251 | - | select { | 292 | + | wg.Add(1) |
| 252 | - | case <-ext.readyCh: | 293 | + | go func(ext *Extension) { |
| 253 | - | case <-time.After(grace): | 294 | + | defer wg.Done() |
| 254 | - | fmt.Fprintf(ext.logFile, "[zot] timed out waiting for ready frame; proceeding\n") | 295 | + | select { |
| 255 | - | ext.readyOnce.Do(func() { close(ext.readyCh) }) | 296 | + | case <-ext.readyCh: |
| 256 | - | } | 297 | + | case <-deadline: |
| 298 | + | fmt.Fprintf(ext.logFile, "[zot] timed out waiting for ready frame; proceeding\n") | |||
| 299 | + | ext.readyOnce.Do(func() { close(ext.readyCh) }) | |||
| 300 | + | } | |||
| 301 | + | }(ext) | |||
| 257 | } | 302 | } | ||
| 303 | + | wg.Wait() | |||
| 258 | } | 304 | } | ||
| 259 | 305 | ||||
| 260 | // spawn launches the subprocess, hooks up pipes, logs stderr, and | 306 | // spawn launches the subprocess, hooks up pipes, logs stderr, and | ||
| @@ -344,9 +390,48 @@ func (m *Manager) spawn(ctx context.Context, ext *Extension) error { | |||||
| 344 | 390 | ||||
| 345 | // Spin up the read loop now that the handshake is done. | 391 | // Spin up the read loop now that the handshake is done. | ||
| 346 | go m.readLoop(ext, scanner) | 392 | go m.readLoop(ext, scanner) | ||
| 393 | + | ||||
| 394 | + | // Compatibility shim: extensions built against the phase-1 SDK | |||
| 395 | + | // don't send a ready frame. Watch the read loop's frame arrival | |||
| 396 | + | // rate; if nothing's arrived for readyIdleWindow we treat the | |||
| 397 | + | // extension as ready so WaitForReady doesn't burn the full grace | |||
| 398 | + | // on every startup. Newer extensions still trigger the explicit | |||
| 399 | + | // path on their own ready frame. | |||
| 400 | + | go m.assumeReadyAfterIdle(ext) | |||
| 401 | + | ||||
| 347 | return nil | 402 | return nil | ||
| 348 | } | 403 | } | ||
| 349 | 404 | ||||
| 405 | + | // readyIdleWindow is how long the manager waits for a frame after | |||
| 406 | + | // hello before assuming an extension that doesn't send `ready` is | |||
| 407 | + | // nevertheless ready. 250ms is enough for any well-behaved native | |||
| 408 | + | // binary to flush its register frames; slow runtimes (npx tsx) flush | |||
| 409 | + | // even faster once they've started, so this rarely affects them. | |||
| 410 | + | const readyIdleWindow = 250 * time.Millisecond | |||
| 411 | + | ||||
| 412 | + | func (m *Manager) assumeReadyAfterIdle(ext *Extension) { | |||
| 413 | + | last := ext.lastFrameTime | |||
| 414 | + | for { | |||
| 415 | + | select { | |||
| 416 | + | case <-ext.readyCh: | |||
| 417 | + | return | |||
| 418 | + | case <-time.After(readyIdleWindow): | |||
| 419 | + | } | |||
| 420 | + | ext.mu.Lock() | |||
| 421 | + | current := ext.lastFrameTime | |||
| 422 | + | ext.mu.Unlock() | |||
| 423 | + | if current.Equal(last) { | |||
| 424 | + | // No new frame in the idle window. Treat as ready. | |||
| 425 | + | ext.readyOnce.Do(func() { | |||
| 426 | + | fmt.Fprintf(ext.logFile, "[zot] no ready frame; auto-readying after idle (legacy SDK?)\n") | |||
| 427 | + | close(ext.readyCh) | |||
| 428 | + | }) | |||
| 429 | + | return | |||
| 430 | + | } | |||
| 431 | + | last = current | |||
| 432 | + | } | |||
| 433 | + | } | |||
| 434 | + | ||||
| 350 | // readLoop processes every frame the extension sends after hello. | 435 | // readLoop processes every frame the extension sends after hello. | ||
| 351 | // Returns when stdout closes. | 436 | // Returns when stdout closes. | ||
| 352 | func (m *Manager) readLoop(ext *Extension, scanner *bufio.Scanner) { | 437 | func (m *Manager) readLoop(ext *Extension, scanner *bufio.Scanner) { | ||
| @@ -372,6 +457,9 @@ func (m *Manager) readLoop(ext *Extension, scanner *bufio.Scanner) { | |||||
| 372 | 457 | ||||
| 373 | for scanner.Scan() { | 458 | for scanner.Scan() { | ||
| 374 | line := scanner.Bytes() | 459 | line := scanner.Bytes() | ||
| 460 | + | ext.mu.Lock() | |||
| 461 | + | ext.lastFrameTime = time.Now() | |||
| 462 | + | ext.mu.Unlock() | |||
| 375 | var frame extproto.Frame | 463 | var frame extproto.Frame | ||
| 376 | if err := json.Unmarshal(line, &frame); err != nil { | 464 | if err := json.Unmarshal(line, &frame); err != nil { | ||
| 377 | fmt.Fprintf(ext.logFile, "[zot] malformed json from extension: %v\n", err) | 465 | fmt.Fprintf(ext.logFile, "[zot] malformed json from extension: %v\n", err) | ||