MCPcopy Create free account
hub / github.com/PostHog/duckgres / SpawnWorker

Method SpawnWorker

controlplane/worker_mgr.go:349–513  ·  view source on GitHub ↗

SpawnWorker starts a new duckdb-service worker process. It uses a pre-bound socket from the pool if available, falling back to binding a new socket (which may fail with EROFS under systemd's ProtectSystem=strict after startup).

(id int)

Source from the content-addressed store, hash-verified

347 slog.Warn("Primary socket directory is read-only, falling back.", "primary", p.socketDir, "fallback", fallback)
348 p.preboundMu.Lock()
349 p.fallbackSocketDir = fallback
350 p.preboundMu.Unlock()
351 return fallback, nil
352}
353
354// SpawnWorker starts a new duckdb-service worker process.
355// It uses a pre-bound socket from the pool if available, falling back to
356// binding a new socket (which may fail with EROFS under systemd's
357// ProtectSystem=strict after startup).
358func (p *FlightWorkerPool) SpawnWorker(id int) error {
359 spawnStart := time.Now()
360 defer func() {
361 observeControlPlaneWorkerSpawn(time.Since(spawnStart))
362 }()
363
364 token := generateToken()
365
366 // Try to use a pre-bound socket first. These are bound eagerly at startup
367 // while the socket directory is verified writable, avoiding EROFS errors
368 // that can occur later under systemd ProtectSystem=strict.
369 ps := p.takePrebound()
370
371 var ln net.Listener
372 var socketPath string
373 if ps != nil {
374 ln = ps.listener
375 socketPath = ps.socketPath
376 } else {
377 // Fallback: bind a new socket. This can happen when workers are being
378 // retired asynchronously and their pre-bound sockets haven't been
379 // returned to the pool yet. Uses effectiveSocketDir() to fall back
380 // to /tmp/duckgres if the primary dir went read-only (EROFS).
381 dir, err := p.effectiveSocketDir()
382 if err != nil {
383 return fmt.Errorf("no writable socket directory: %w", err)
384 }
385 socketPath = fmt.Sprintf("%s/worker-dyn-%d.sock", dir, id)
386 _ = os.Remove(socketPath)
387 var listenErr error
388 ln, listenErr = net.Listen("unix", socketPath)
389 if listenErr != nil {
390 return fmt.Errorf("bind worker socket %s: %w", socketPath, listenErr)
391 }
392 if err := os.Chmod(socketPath, 0700); err != nil {
393 slog.Warn("Failed to set worker socket permissions.", "error", err)
394 }
395 }
396
397 // Get a dup'd FD to pass to the child. ExtraFiles[0] becomes FD 3.
398 file, err := ln.(*net.UnixListener).File()
399 if err != nil {
400 if ps != nil {
401 p.returnPrebound(ps)
402 } else {
403 _ = ln.Close()
404 }
405 return fmt.Errorf("get listener fd for worker %d: %w", id, err)
406 }

Callers 3

SpawnAllMethod · 0.95
AcquireWorkerMethod · 0.95

Calls 12

takePreboundMethod · 0.95
effectiveSocketDirMethod · 0.95
returnPreboundMethod · 0.95
generateTokenFunction · 0.85
appendOrReplaceEnvFunction · 0.85
waitForWorkerFunction · 0.85
NowMethod · 0.80
RemoveMethod · 0.80
CloseMethod · 0.65
StartMethod · 0.45

Tested by 1