diff --git a/cmd/relayfile-cli/commandspec.go b/cmd/relayfile-cli/commandspec.go index 8dd8ab3e..332c45d6 100644 --- a/cmd/relayfile-cli/commandspec.go +++ b/cmd/relayfile-cli/commandspec.go @@ -936,6 +936,7 @@ func mountOptions() []cliOptionSpec { {Flags: "--mode ", Description: "mount mode: poll (recommended) or fuse"}, {Flags: "--interval ", Description: "sync interval"}, {Flags: "--interval-jitter ", Description: "sync interval jitter ratio (0.0-1.0)"}, + {Flags: "--startup-jitter ", Description: "maximum random delay before the first sync cycle, so mounts started together (scheduled sandboxes, scoped siblings) do not bootstrap in lockstep (0 disables)"}, {Flags: "--timeout ", Description: "per-sync timeout"}, {Flags: "--bootstrap-timeout ", Description: "hard cap for the one-time/full-tree bootstrap pull (0 = unbounded while making progress)"}, {Flags: "--bootstrap-max-files-per-cycle ", Description: "maximum files materialized per resumable tree-bootstrap cycle (-1 = legacy unbounded tree behavior)"}, diff --git a/cmd/relayfile-cli/main.go b/cmd/relayfile-cli/main.go index 6c148648..2f84f217 100644 --- a/cmd/relayfile-cli/main.go +++ b/cmd/relayfile-cli/main.go @@ -15,6 +15,7 @@ import ( "fmt" "io" "log" + "math" mathrand "math/rand/v2" "mime" "net/http" @@ -7357,6 +7358,7 @@ func runMount(args []string) error { mode := fs.String("mode", envOrDefault("RELAYFILE_MOUNT_MODE", defaultMountMode), "mount mode: poll (recommended) or fuse") interval := fs.Duration("interval", durationEnv("RELAYFILE_MOUNT_INTERVAL", defaultMountInterval), "sync interval") intervalJitter := fs.Float64("interval-jitter", floatEnv("RELAYFILE_MOUNT_INTERVAL_JITTER", 0.2), "sync interval jitter ratio (0.0-1.0)") + startupJitter := fs.Duration("startup-jitter", durationEnv("RELAYFILE_MOUNT_STARTUP_JITTER", defaultStartupJitter), "maximum random delay before the first sync cycle, so mounts started together (scheduled sandboxes, scoped siblings) do not bootstrap in lockstep (0 disables)") timeout := fs.Duration("timeout", durationEnv("RELAYFILE_MOUNT_TIMEOUT", defaultMountTimeout), "per-sync timeout") bootstrapTimeout := fs.Duration("bootstrap-timeout", durationEnv("RELAYFILE_BOOTSTRAP_TIMEOUT", 0), "hard cap for the one-time/full-tree bootstrap pull (0 = unbounded while making progress)") bootstrapMaxFiles := fs.Int("bootstrap-max-files-per-cycle", intEnv("RELAYFILE_BOOTSTRAP_MAX_FILES_PER_CYCLE", 2000), "maximum files materialized per resumable tree-bootstrap cycle (-1 = legacy unbounded tree behavior)") @@ -7820,6 +7822,7 @@ func runMount(args []string) error { *timeout = defaultMountTimeout } *intervalJitter = clampJitterRatio(*intervalJitter) + *startupJitter = clampStartupJitter(*startupJitter) registerPID := shouldRegisterMountPID(*daemonized, *once) if *daemonized { @@ -7935,6 +7938,7 @@ func runMount(args []string) error { *timeout, *interval, *intervalJitter, + *startupJitter, *websocketEnabled, *once, *daemonized, @@ -14095,6 +14099,42 @@ func boolPtr(value bool) *bool { return &value } +// defaultStartupJitter spreads the first (possibly full-tree) sync of mounts +// that start in the same instant — scheduled sandboxes at a cron boundary, and +// every scoped runner of one mount — so they do not all hit the +// single-threaded workspace Durable Object in the same second. Mirrors +// cmd/relayfile-mount. +const ( + defaultStartupJitter = 5 * time.Second + maxStartupJitter = 5 * time.Minute +) + +// startupSplaySample draws the uniform sample for the startup splay; tests pin +// it so a wait can be asserted without depending on a random draw. +var startupSplaySample = mathrand.Float64 + +func clampStartupJitter(value time.Duration) time.Duration { + if value < 0 { + return 0 + } + if value > maxStartupJitter { + return maxStartupJitter + } + return value +} + +// startupSplayDelay maps a uniform sample in [0,1) onto [0, max). +func startupSplayDelay(max time.Duration, sample float64) time.Duration { + max = clampStartupJitter(max) + if max == 0 || sample <= 0 { + return 0 + } + if sample >= 1 { + sample = math.Nextafter(1, 0) + } + return time.Duration(sample * float64(max)) +} + func clampJitterRatio(value float64) float64 { if value < 0 { return 0 @@ -14343,6 +14383,7 @@ func runMountLoop(rootCtx context.Context, syncer *mountsync.Syncer, localDir, w timeout, interval, intervalJitter, + 0, websocketEnabled, once, daemonized, @@ -14352,7 +14393,7 @@ func runMountLoop(rootCtx context.Context, syncer *mountsync.Syncer, localDir, w ) } -func runMountLoopWithAuthLock(rootCtx context.Context, syncer *mountsync.Syncer, localDir, workspaceID, serverURL, delegatedCredsFile string, timeout, interval time.Duration, intervalJitter float64, websocketEnabled, once, daemonized bool, pidFile, logFile string, authMu *sync.Mutex) error { +func runMountLoopWithAuthLock(rootCtx context.Context, syncer *mountsync.Syncer, localDir, workspaceID, serverURL, delegatedCredsFile string, timeout, interval time.Duration, intervalJitter float64, startupJitter time.Duration, websocketEnabled, once, daemonized bool, pidFile, logFile string, authMu *sync.Mutex) error { interval = enforcePollIntervalFloor(interval) httpClient, _ := syncerClient(syncer) record, _ := workspaceRecordByID(workspaceID) @@ -14719,6 +14760,23 @@ func runMountLoopWithAuthLock(rootCtx context.Context, syncer *mountsync.Syncer, } log.Print(mountStartBanner(localDir, interval, intervalJitter)) + // Each runner (one per scope) draws its own splay. The watcher is already + // admitting local edits, so only the first remote cycle waits. + if delay := startupSplayDelay(startupJitter, startupSplaySample()); delay > 0 { + log.Printf("mount startup splay: waiting %s before first sync", delay.Round(time.Millisecond)) + splay := time.NewTimer(delay) + select { + case <-splay.C: + case <-rootCtx.Done(): + splay.Stop() + if once { + return fmt.Errorf("initial sync cancelled before first cycle: %w", rootCtx.Err()) + } + log.Printf("mount sync stopping: %v", rootCtx.Err()) + writeSnapshot() + return nil + } + } initialErr := runCycle(true) logStuckEventSummary(syncer, initialErr) if mountsync.IsBootstrapTerminalError(initialErr) { diff --git a/cmd/relayfile-cli/startup_splay_test.go b/cmd/relayfile-cli/startup_splay_test.go new file mode 100644 index 00000000..fb9c4484 --- /dev/null +++ b/cmd/relayfile-cli/startup_splay_test.go @@ -0,0 +1,77 @@ +package main + +import ( + "context" + "io" + "log" + "net/http" + "net/http/httptest" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/agentworkforce/relayfile/internal/mountsync" +) + +// `relayfile mount` runs its own loop, not relayfile-mount's. Its first +// (possibly full-tree) cycle must also wait out the startup splay, or every +// scheduled sandbox and every scoped runner bootstraps in the same second. +func TestPublicMountLoopWaitsStartupSplayBeforeFirstRequest(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + clearRelayfileEnv(t) + var requests atomic.Int64 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + switch { + case strings.Contains(r.URL.Path, "/fs/tree"): + _, _ = w.Write([]byte(`{"path":"/","entries":[]}`)) + case strings.Contains(r.URL.Path, "/fs/events"): + _, _ = w.Write([]byte(`{"events":[]}`)) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + previous := startupSplaySample + startupSplaySample = func() float64 { return 0.5 } // 2.5m of a 5m window + t.Cleanup(func() { startupSplaySample = previous }) + prevLog := log.Writer() + log.SetOutput(io.Discard) + defer log.SetOutput(prevLog) + + localDir := filepath.Join(t.TempDir(), "mount") + ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond) + defer cancel() + disableWebSocket := false + syncer, err := mountsync.NewSyncer(mountsync.NewHTTPClient(server.URL, "test-token", server.Client()), mountsync.SyncerOptions{ + WorkspaceID: "ws_public_startup_splay", RemoteRoot: "/", LocalRoot: localDir, + Interval: time.Hour, WebSocket: &disableWebSocket, RootCtx: ctx, Logger: log.New(io.Discard, "", 0), + }) + if err != nil { + t.Fatalf("NewSyncer: %v", err) + } + err = runMountLoopWithAuthLock(ctx, syncer, localDir, "ws_public_startup_splay", server.URL, "", + time.Second, time.Hour, 0, maxStartupJitter, false, true, false, + mountPIDFile(localDir), mountLogFile(localDir), &sync.Mutex{}) + if err == nil { + t.Fatal("one-shot public mount returned success while its startup splay outlasted the deadline") + } + if got := requests.Load(); got != 0 { + t.Fatalf("public mount made %d request(s) during the startup splay, want 0", got) + } +} + +func TestPublicMountStartupSplayDelayBounds(t *testing.T) { + if got := startupSplayDelay(0, 0.9); got != 0 { + t.Fatalf("disabled splay = %s, want 0", got) + } + if got := startupSplayDelay(10*time.Second, 0.5); got != 5*time.Second { + t.Fatalf("half sample = %s, want 5s", got) + } + if got := startupSplayDelay(time.Hour, 0.999); got >= maxStartupJitter { + t.Fatalf("oversized splay = %s, want < %s", got, maxStartupJitter) + } +} diff --git a/cmd/relayfile-mount/main.go b/cmd/relayfile-mount/main.go index ea7a8545..9417ff34 100644 --- a/cmd/relayfile-mount/main.go +++ b/cmd/relayfile-mount/main.go @@ -8,6 +8,7 @@ import ( "flag" "fmt" "log" + "math" "math/rand" "net/http" "os" @@ -66,6 +67,7 @@ type mountConfig struct { syncMode string interval time.Duration intervalJitter float64 + startupJitter time.Duration timeout time.Duration bootstrapTimeout time.Duration bootstrapMaxFiles int @@ -122,6 +124,7 @@ func main() { syncModeFlag := flag.String("sync-mode", envOrDefault("RELAYFILE_MOUNT_SYNC_MODE", syncModeMirror), "sync behavior: mirror (pull and push), pull-only (poll mode only; mirror remote changes without writeback), or write-only (push local changes without mirroring provider history)") interval := flag.Duration("interval", durationEnv("RELAYFILE_MOUNT_INTERVAL", 30*time.Second), "sync interval") intervalJitter := flag.Float64("interval-jitter", floatEnv("RELAYFILE_MOUNT_INTERVAL_JITTER", 0.2), "sync interval jitter ratio (0.0-1.0)") + startupJitter := flag.Duration("startup-jitter", durationEnv("RELAYFILE_MOUNT_STARTUP_JITTER", defaultStartupJitter), "maximum random delay before the first sync cycle, so mounts started together (scheduled sandboxes, scoped siblings) do not bootstrap in lockstep (0 disables)") timeout := flag.Duration("timeout", durationEnv("RELAYFILE_MOUNT_TIMEOUT", 15*time.Second), "per-sync timeout") bootstrapTimeout := flag.Duration("bootstrap-timeout", durationEnv("RELAYFILE_BOOTSTRAP_TIMEOUT", 0), "hard cap for the one-time/full-tree bootstrap pull (0 = unbounded while making progress)") bootstrapMaxFiles := flag.Int("bootstrap-max-files-per-cycle", intEnv("RELAYFILE_BOOTSTRAP_MAX_FILES_PER_CYCLE", 2000), "maximum files materialized per resumable tree-bootstrap cycle (-1 = legacy unbounded tree behavior)") @@ -193,6 +196,7 @@ func main() { } allRemotePaths := append(remotePaths.Values(), fileRemotePaths...) *intervalJitter = clampJitterRatio(*intervalJitter) + *startupJitter = clampStartupJitter(*startupJitter) resolvedMode, err := resolveMountMode(*mode, *fuse) if err != nil { log.Fatalf("invalid mount mode: %v", err) @@ -233,6 +237,7 @@ func main() { syncMode: resolvedSyncMode, interval: *interval, intervalJitter: *intervalJitter, + startupJitter: *startupJitter, timeout: *timeout, bootstrapTimeout: *bootstrapTimeout, bootstrapMaxFiles: *bootstrapMaxFiles, @@ -727,7 +732,23 @@ func runSinglePollingMount(rootCtx context.Context, cfg mountConfig) error { // "just finished as part of this attempt." priorBootstrapComplete := bootstrapAlreadyComplete(cfg.localDir) - if err := run(true); err != nil { + flushedDuringSplay, err := waitStartupSplay(rootCtx, startupSplayDelay(cfg.startupJitter, startupSplaySample()), cfg.flushReq) + if err != nil { + if cfg.once { + return fmt.Errorf("initial sync cancelled before first cycle: %w", err) + } + log.Printf("mount sync stopping: %v", err) + return nil + } + if flushedDuringSplay { + // An explicit flush ends the splay: the operator asked for a sync + // now, and the notifier stops waiting for an ack long before a + // full-length splay would elapse. The kicked reconcile is the first + // cycle. + if err := serviceFlushRequest(rootCtx, cfg, syncer); err != nil { + return err + } + } else if err := run(true); err != nil { return err } if cfg.once { @@ -791,16 +812,8 @@ func runSinglePollingMount(rootCtx context.Context, cfg mountConfig) error { log.Printf("mount sync stopping: %v", rootCtx.Err()) return nil case <-cfg.flushReq: - kickErr := kickReconcile(rootCtx, cfg, syncer) - if recErr := recordFlushAck(cfg, kickErr); recErr != nil { - log.Printf("mount flush ack failed: %v", recErr) - } else if kickErr != nil { - log.Printf("mount flush requested via SIGUSR1; failed: %v", kickErr) - } else { - log.Printf("mount flush requested via SIGUSR1; ack recorded") - } - if kickErr != nil && mountsync.IsBootstrapTerminalError(kickErr) { - return kickErr + if err := serviceFlushRequest(rootCtx, cfg, syncer); err != nil { + return err } case <-wsTicker.C: if mountWebSocketEnabled(cfg) { @@ -1532,6 +1545,62 @@ func mountReconcileUsesWebSocketCadence(cfg mountConfig, watcherActive bool) boo return mountWebSocketEnabled(cfg) && (cfg.syncMode == syncModePullOnly || watcherActive) } +// defaultStartupJitter spreads the first (possibly full-tree) sync of mounts +// that start in the same instant. Scheduled sandboxes launch together at cron +// boundaries and every scoped sibling starts its own Syncer at once; without a +// splay they all hit the single-threaded workspace Durable Object in the same +// second. Five seconds is small against the bootstrap readiness budget. +const ( + defaultStartupJitter = 5 * time.Second + maxStartupJitter = 5 * time.Minute +) + +// startupSplaySample draws the uniform sample for the startup splay; tests pin +// it so a wait can be asserted without depending on a random draw. +var startupSplaySample = rand.Float64 + +func clampStartupJitter(value time.Duration) time.Duration { + if value < 0 { + return 0 + } + if value > maxStartupJitter { + return maxStartupJitter + } + return value +} + +// startupSplayDelay maps a uniform sample in [0,1) onto [0, max). +func startupSplayDelay(max time.Duration, sample float64) time.Duration { + max = clampStartupJitter(max) + if max == 0 || sample <= 0 { + return 0 + } + if sample >= 1 { + sample = math.Nextafter(1, 0) + } + return time.Duration(sample * float64(max)) +} + +// waitStartupSplay waits out the splay unless ctx ends or a flush request +// arrives first; flushed reports the latter so the caller services it. +func waitStartupSplay(ctx context.Context, delay time.Duration, flushReq <-chan struct{}) (flushed bool, err error) { + if delay <= 0 { + return false, ctx.Err() + } + log.Printf("mount startup splay: waiting %s before first sync", delay.Round(time.Millisecond)) + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-timer.C: + return false, nil + case <-flushReq: + log.Printf("mount startup splay: flush requested; starting first sync now") + return true, nil + case <-ctx.Done(): + return false, ctx.Err() + } +} + func clampJitterRatio(value float64) float64 { if value < 0 { return 0 diff --git a/cmd/relayfile-mount/notify_flush.go b/cmd/relayfile-mount/notify_flush.go index a7a6e953..f4b75c26 100644 --- a/cmd/relayfile-mount/notify_flush.go +++ b/cmd/relayfile-mount/notify_flush.go @@ -2,6 +2,7 @@ package main import ( "context" + "log" "os" "time" @@ -15,6 +16,24 @@ func kickReconcile(rootCtx context.Context, cfg mountConfig, syncer *mountsync.S return syncer.Reconcile(ctx) } +// serviceFlushRequest runs the reconcile a SIGUSR1 flush asked for and +// records its ack. Only a terminal bootstrap error is returned; any other +// failure is reported through the ack. +func serviceFlushRequest(rootCtx context.Context, cfg mountConfig, syncer *mountsync.Syncer) error { + kickErr := kickReconcile(rootCtx, cfg, syncer) + if recErr := recordFlushAck(cfg, kickErr); recErr != nil { + log.Printf("mount flush ack failed: %v", recErr) + } else if kickErr != nil { + log.Printf("mount flush requested via SIGUSR1; failed: %v", kickErr) + } else { + log.Printf("mount flush requested via SIGUSR1; ack recorded") + } + if kickErr != nil && mountsync.IsBootstrapTerminalError(kickErr) { + return kickErr + } + return nil +} + func recordFlushAck(cfg mountConfig, kickErr error) error { prev, err := mountlease.ReadFlushAck(cfg.baseURL, cfg.workspaceID, cfg.localDir) if err != nil { diff --git a/cmd/relayfile-mount/startup_splay_test.go b/cmd/relayfile-mount/startup_splay_test.go new file mode 100644 index 00000000..73e954ee --- /dev/null +++ b/cmd/relayfile-mount/startup_splay_test.go @@ -0,0 +1,153 @@ +package main + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/agentworkforce/relayfile/internal/mountlease" + "github.com/agentworkforce/relayfile/internal/mountsync" +) + +func TestStartupSplayDelayBounds(t *testing.T) { + if got := startupSplayDelay(0, 0.9); got != 0 { + t.Fatalf("disabled splay = %s, want 0", got) + } + if got := startupSplayDelay(-time.Second, 0.9); got != 0 { + t.Fatalf("negative splay = %s, want 0", got) + } + if got := startupSplayDelay(10*time.Second, 0.5); got != 5*time.Second { + t.Fatalf("half sample = %s, want 5s", got) + } + if got := startupSplayDelay(10*time.Second, 1); got >= 10*time.Second { + t.Fatalf("sample 1 = %s, want < 10s", got) + } + if got := startupSplayDelay(time.Hour, 0.999); got >= maxStartupJitter { + t.Fatalf("oversized splay = %s, want < %s", got, maxStartupJitter) + } +} + +// The first cycle of a mount (a full-tree bootstrap for a fresh sandbox) must +// wait out the startup splay before touching the workspace, so mounts +// launched in the same instant do not bootstrap in lockstep. +func TestRunSinglePollingMountWaitsStartupSplayBeforeFirstRequest(t *testing.T) { + var requests atomic.Int64 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + switch { + case strings.Contains(r.URL.Path, "/fs/tree"): + _, _ = w.Write([]byte(`{"path":"/","entries":[]}`)) + case strings.Contains(r.URL.Path, "/fs/events"): + _, _ = w.Write([]byte(`{"events":[]}`)) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + + cfg := mountConfig{ + baseURL: server.URL, + token: "test-token", + workspaceID: "ws_startup_splay", + remotePath: "/", + localDir: t.TempDir(), + stateDir: t.TempDir(), + mountKind: mountsync.MountKindDaemon, + syncMode: syncModeMirror, + interval: time.Hour, + timeout: time.Second, + startupJitter: maxStartupJitter, + once: true, + } + // A pinned sample (2.5m of a 5m window) outlasts the 150ms deadline, so + // the mount must still be waiting when the context expires. + pinStartupSplaySample(t, 0.5) + ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond) + defer cancel() + err := runSinglePollingMount(ctx, cfg) + if err == nil { + t.Fatal("one-shot mount returned success while its startup splay outlasted the deadline") + } + if got := requests.Load(); got != 0 { + t.Fatalf("mount made %d request(s) during the startup splay, want 0", got) + } +} + +// pinStartupSplaySample fixes the splay draw for one test. +func pinStartupSplaySample(t *testing.T, sample float64) { + t.Helper() + previous := startupSplaySample + startupSplaySample = func() float64 { return sample } + t.Cleanup(func() { startupSplaySample = previous }) +} + +func TestWaitStartupSplayZeroDelayRunsImmediately(t *testing.T) { + start := time.Now() + if _, err := waitStartupSplay(context.Background(), 0, nil); err != nil { + t.Fatalf("zero splay: %v", err) + } + if elapsed := time.Since(start); elapsed > 50*time.Millisecond { + t.Fatalf("zero splay waited %s", elapsed) + } +} + +// A `--notify-flush` against a daemon still inside its startup splay must be +// serviced, not left queued until the splay ends: the notifier gives up after +// notifyFlushWait (2x the cycle timeout), far below the up-to-5m splay. +func TestFlushRequestDuringStartupSplayIsAcknowledgedPromptly(t *testing.T) { + cacheHome := t.TempDir() + t.Setenv("HOME", cacheHome) + t.Setenv("XDG_CACHE_HOME", cacheHome) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.Contains(r.URL.Path, "/fs/tree"): + _, _ = w.Write([]byte(`{"path":"/","entries":[]}`)) + case strings.Contains(r.URL.Path, "/fs/events"): + _, _ = w.Write([]byte(`{"events":[]}`)) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + pinStartupSplaySample(t, 0.5) // 2.5m splay: far past this test's deadline + + flushReq := make(chan struct{}, 1) + localDir := t.TempDir() + cfg := mountConfig{ + baseURL: server.URL, + token: "test-token", + workspaceID: "ws_startup_splay_flush", + remotePath: "/", + localDir: localDir, + stateDir: t.TempDir(), + mountKind: mountsync.MountKindDaemon, + mode: mountModePoll, + syncMode: syncModeMirror, + interval: time.Hour, + timeout: 2 * time.Second, + startupJitter: maxStartupJitter, + flushReq: flushReq, + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- runSinglePollingMount(ctx, cfg) }() + defer func() { cancel(); <-done }() + + flushReq <- struct{}{} + deadline := time.Now().Add(notifyFlushWait(cfg)) + for time.Now().Before(deadline) { + ack, err := mountlease.ReadFlushAck(cfg.baseURL, cfg.workspaceID, localDir) + if err == nil && ack.Seq > 0 { + if !ack.OK { + t.Fatalf("flush ack recorded a failure: %s", ack.Error) + } + return + } + time.Sleep(20 * time.Millisecond) + } + t.Fatalf("flush request during the startup splay was not acknowledged within %s", notifyFlushWait(cfg)) +} diff --git a/internal/mountsync/fullpull_gate.go b/internal/mountsync/fullpull_gate.go new file mode 100644 index 00000000..51e62dd5 --- /dev/null +++ b/internal/mountsync/fullpull_gate.go @@ -0,0 +1,78 @@ +package mountsync + +import ( + "context" + "os" + "strconv" + "strings" + "sync" +) + +// defaultFullPullReadConcurrency caps in-flight full-tree remote reads (tree +// pages, bulk reads, bootstrap point reads, export snapshots, GitHub clone +// manifests and tar seed streams) across EVERY +// Syncer in this process. A scoped mount runs one Syncer per remote path, and +// each Syncer already bounds itself to defaultBootstrapReadWorkers; without a +// process-wide gate, N scopes bootstrapping together multiplied that to 4N +// concurrent reads against the same single-threaded workspace Durable Object. +// Four keeps a single-scope mount at its historical throughput. +const ( + defaultFullPullReadConcurrency = 4 + maxFullPullReadConcurrency = 16 +) + +// fullPullReadGate is process-wide on purpose: the resource it protects is the +// remote workspace, not one Syncer's state. +var fullPullReadGate = newReadGate(fullPullReadConcurrency()) + +type readGate struct { + slots chan struct{} +} + +func newReadGate(limit int) *readGate { + if limit <= 0 { + limit = defaultFullPullReadConcurrency + } + return &readGate{slots: make(chan struct{}, limit)} +} + +// do runs fn while holding one slot. Waiting for a slot honours ctx so a +// cancelled or timed-out cycle never blocks behind siblings; the caller sees +// ctx.Err() exactly as if the request itself had been cancelled, which the +// bootstrap runner already treats as a resumable yield. +func (g *readGate) do(ctx context.Context, fn func() error) error { + release, err := g.acquire(ctx) + if err != nil { + return err + } + defer release() + return fn() +} + +// acquire takes one slot for a read whose lifetime outlives a single call — a +// streamed response that is consumed after the request returns. The caller +// must call release exactly once, when the stream is closed. +func (g *readGate) acquire(ctx context.Context) (release func(), err error) { + select { + case g.slots <- struct{}{}: + case <-ctx.Done(): + return nil, ctx.Err() + } + var once sync.Once + return func() { once.Do(func() { <-g.slots }) }, nil +} + +func fullPullReadConcurrency() int { + raw := strings.TrimSpace(os.Getenv("RELAYFILE_FULL_PULL_READ_CONCURRENCY")) + if raw == "" { + return defaultFullPullReadConcurrency + } + v, err := strconv.Atoi(raw) + if err != nil || v <= 0 { + return defaultFullPullReadConcurrency + } + if v > maxFullPullReadConcurrency { + return maxFullPullReadConcurrency + } + return v +} diff --git a/internal/mountsync/fullpull_gate_test.go b/internal/mountsync/fullpull_gate_test.go new file mode 100644 index 00000000..f6ea5a61 --- /dev/null +++ b/internal/mountsync/fullpull_gate_test.go @@ -0,0 +1,303 @@ +package mountsync + +import ( + "context" + "io" + "sync" + "sync/atomic" + "testing" + "time" +) + +// Scoped mounts run one Syncer per remote path in the same process. Their +// initial full pulls must share one read budget against the workspace instead +// of each bringing its own worker pool (the :00 bootstrap stampede that pushed +// the workspace Durable Object past its memory limit). +func TestConcurrentSyncerBootstrapsShareProcessWideReadBudget(t *testing.T) { + t.Setenv("RELAYFILE_BOOTSTRAP_READ_CONCURRENCY", "4") + withFullPullReadGate(t, defaultFullPullReadConcurrency) + client := newBootstrapClient(24, 24) + client.readFileSleep = 20 * time.Millisecond + + const scopes = 3 + syncers := make([]*Syncer, scopes) + for i := range syncers { + syncers[i] = newBootstrapSyncer(t, client, t.TempDir(), SyncerOptions{RootCtx: context.Background()}) + } + + var wg sync.WaitGroup + errs := make(chan error, scopes) + for _, s := range syncers { + s := s + wg.Add(1) + go func() { + defer wg.Done() + errs <- s.Reconcile(context.Background()) + }() + } + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatalf("reconcile failed: %v", err) + } + } + + if got, limit := client.maxActiveRead.Load(), int64(defaultFullPullReadConcurrency); got > limit { + t.Fatalf("max concurrent bootstrap reads across %d syncers = %d, want <= %d", scopes, got, limit) + } + if got, want := client.readFileCalls.Load(), int64(24*scopes); got != want { + t.Fatalf("expected every scope to converge with %d reads, got %d", want, got) + } +} + +func TestReadGateWaitHonoursContext(t *testing.T) { + gate := newReadGate(1) + release := make(chan struct{}) + holding := make(chan struct{}) + go func() { + _ = gate.do(context.Background(), func() error { + close(holding) + <-release + return nil + }) + }() + <-holding + defer close(release) + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + called := false + err := gate.do(ctx, func() error { called = true; return nil }) + if err != context.DeadlineExceeded { + t.Fatalf("gate wait error = %v, want context.DeadlineExceeded", err) + } + if called { + t.Fatal("gated fn ran without a slot") + } +} + +func TestFullPullReadConcurrencyEnv(t *testing.T) { + for raw, want := range map[string]int{"": 4, "2": 2, "0": 4, "bogus": 4, "99": maxFullPullReadConcurrency} { + t.Setenv("RELAYFILE_FULL_PULL_READ_CONCURRENCY", raw) + if got := fullPullReadConcurrency(); got != want { + t.Fatalf("RELAYFILE_FULL_PULL_READ_CONCURRENCY=%q -> %d, want %d", raw, got, want) + } + } +} + +// withFullPullReadGate installs a gate of exactly limit slots for one test, so +// assertions about the cap never depend on RELAYFILE_FULL_PULL_READ_CONCURRENCY +// in the environment that initialised the package gate. +func withFullPullReadGate(t *testing.T, limit int) *readGate { + t.Helper() + previous := fullPullReadGate + gate := newReadGate(limit) + fullPullReadGate = gate + t.Cleanup(func() { fullPullReadGate = previous }) + return gate +} + +// githubSeedGateClient streams its tar export through a body that stays open +// until released, and counts clone-manifest reads, so a test can observe +// whether the GitHub seed holds a full-pull slot for the whole stream. +type githubSeedGateClient struct { + *fakeExportClient + manifestPath string + manifestReads atomic.Int64 + bodyOpen chan struct{} + bodyRelease chan struct{} +} + +func (c *githubSeedGateClient) ReadFile(ctx context.Context, workspaceID, path string) (RemoteFile, error) { + if normalizeRemotePath(path) == normalizeRemotePath(c.manifestPath) { + c.manifestReads.Add(1) + } + return c.fakeExportClient.fakeClient.ReadFile(ctx, workspaceID, path) +} + +func (c *githubSeedGateClient) ExportGithubWorkingTreeTar(ctx context.Context, workspaceID string, seed GithubWorkingTreeSeedRequest) (GithubWorkingTreeTar, error) { + tarBody, err := c.fakeExportClient.ExportGithubWorkingTreeTar(ctx, workspaceID, seed) + if err != nil { + return tarBody, err + } + close(c.bodyOpen) + tarBody.Body = &heldBody{ReadCloser: tarBody.Body, release: c.bodyRelease} + return tarBody, nil +} + +// heldBody blocks its first Read until released: an export whose response +// headers have arrived but whose (large) body is still streaming. +type heldBody struct { + io.ReadCloser + release <-chan struct{} + once sync.Once +} + +func (b *heldBody) Read(p []byte) (int, error) { + b.once.Do(func() { <-b.release }) + return b.ReadCloser.Read(p) +} + +// The GitHub tar seed is the largest full-pull read there is. Its clone +// manifest read and the whole tar stream — request until body close — must +// sit inside the process-wide full-pull budget, or N GitHub scopes seeding at +// once add N unbounded streams on top of the capped tree/bulk reads. +func TestGithubTarSeedHoldsFullPullSlotUntilBodyClosed(t *testing.T) { + const ( + contentsRoot = "/github/repos/acme/widgets/contents" + headSHA = "head123" + sentinelPath = "/github/repos/acme/widgets/.relayfile/clone.json" + ) + body := []byte("# readme\n") + remotePath := contentsRoot + "/README.md@" + headSHA + ".json" + gate := withFullPullReadGate(t, 1) + client := &githubSeedGateClient{ + fakeExportClient: &fakeExportClient{ + fakeClient: &fakeClient{files: map[string]RemoteFile{ + sentinelPath: {Path: sentinelPath, Revision: "rev_1", ContentType: "application/json", + Content: `{"headSha":"` + headSHA + `","eventsCursor":"evt_seed"}`}, + remotePath: {Path: remotePath, Revision: "rev_2", ContentType: "text/markdown", + Content: string(body), ContentHash: hashBytes(body)}, + }}, + tarFiles: map[string][]byte{"README.md": body}, + }, + manifestPath: sentinelPath, + bodyOpen: make(chan struct{}), + bodyRelease: make(chan struct{}), + } + syncer, err := NewSyncer(client, SyncerOptions{ + WorkspaceID: "ws_github_tar_gate", RemoteRoot: contentsRoot, LocalRoot: t.TempDir(), + WebSocket: boolPtr(false), FullPullEvery: -1, + }) + if err != nil { + t.Fatalf("NewSyncer: %v", err) + } + + // Occupy the only slot: the manifest read must wait for it. + holderRelease := make(chan struct{}) + holding := make(chan struct{}) + go func() { + _ = gate.do(context.Background(), func() error { close(holding); <-holderRelease; return nil }) + }() + <-holding + done := make(chan error, 1) + go func() { done <- syncer.SyncOnce(context.Background()) }() + time.Sleep(100 * time.Millisecond) + if got := client.manifestReads.Load(); got != 0 { + close(holderRelease) + close(client.bodyRelease) + <-done + t.Fatalf("clone manifest read %d time(s) while every full-pull slot was taken, want 0", got) + } + close(holderRelease) + + select { + case <-client.bodyOpen: + case <-time.After(2 * time.Second): + close(client.bodyRelease) + <-done + t.Fatal("github tar export never started") + } + // Headers are back but the body is still streaming: the seed must still + // own its slot. + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + stolen := gate.do(ctx, func() error { return nil }) + cancel() + close(client.bodyRelease) + if err := <-done; err != nil { + t.Fatalf("SyncOnce: %v", err) + } + if stolen == nil { + t.Fatal("a full-pull slot was free while the github tar body was still streaming, want the seed to hold it until close") + } +} + +// githubSeedTreeHookClient runs onTreeListed after each tree page, while the +// page's own gate slot is still held. +type githubSeedTreeHookClient struct { + *githubSeedGateClient + onTreeListed func() +} + +func (c *githubSeedTreeHookClient) ListTree(ctx context.Context, workspaceID, path string, depth int, cursor string) (TreeResponse, error) { + page, err := c.githubSeedGateClient.ListTree(ctx, workspaceID, path, depth, cursor) + if c.onTreeListed != nil { + c.onTreeListed() + } + return page, err +} + +// Waiting for a tar-stream slot must not hold the Syncer mutex: sibling +// scopes can own every slot for a whole stream, and local writeback, outbox +// and watcher handling all need that mutex in the meantime. +func TestGithubTarSeedWaitsForSlotWithoutHoldingSyncerMutex(t *testing.T) { + const ( + contentsRoot = "/github/repos/acme/widgets/contents" + headSHA = "head123" + sentinelPath = "/github/repos/acme/widgets/.relayfile/clone.json" + ) + body := []byte("# readme\n") + remotePath := contentsRoot + "/README.md@" + headSHA + ".json" + gate := withFullPullReadGate(t, 1) + base := &githubSeedGateClient{ + fakeExportClient: &fakeExportClient{ + fakeClient: &fakeClient{files: map[string]RemoteFile{ + sentinelPath: {Path: sentinelPath, Revision: "rev_1", ContentType: "application/json", + Content: `{"headSha":"` + headSHA + `","eventsCursor":"evt_seed"}`}, + remotePath: {Path: remotePath, Revision: "rev_2", ContentType: "text/markdown", + Content: string(body), ContentHash: hashBytes(body)}, + }}, + tarFiles: map[string][]byte{"README.md": body}, + }, + manifestPath: sentinelPath, + bodyOpen: make(chan struct{}), + bodyRelease: make(chan struct{}), + } + close(base.bodyRelease) + // Queue a sibling for the only slot while the tree page still holds it, + // so it wins the slot the moment the page releases and the seed then + // waits for its tar slot. + siblingHolding := make(chan struct{}) + siblingRelease := make(chan struct{}) + var once sync.Once + client := &githubSeedTreeHookClient{githubSeedGateClient: base, onTreeListed: func() { + once.Do(func() { + go func() { + _ = gate.do(context.Background(), func() error { close(siblingHolding); <-siblingRelease; return nil }) + }() + time.Sleep(20 * time.Millisecond) // let the sibling block on the gate first + }) + }} + syncer, err := NewSyncer(client, SyncerOptions{ + WorkspaceID: "ws_github_tar_gate_mutex", RemoteRoot: contentsRoot, LocalRoot: t.TempDir(), + WebSocket: boolPtr(false), FullPullEvery: -1, + }) + if err != nil { + t.Fatalf("NewSyncer: %v", err) + } + done := make(chan error, 1) + go func() { done <- syncer.SyncOnce(context.Background()) }() + select { + case <-siblingHolding: + case <-time.After(2 * time.Second): + close(siblingRelease) + <-done + t.Fatal("sibling never took the slot after the tree snapshot") + } + time.Sleep(50 * time.Millisecond) // the seed is now waiting for its tar slot + + locked := make(chan struct{}) + go func() { syncer.WebSocketConnected(); close(locked) }() + select { + case <-locked: + case <-time.After(500 * time.Millisecond): + close(siblingRelease) + <-done + t.Fatal("Syncer mutex was held while the github tar seed waited for a full-pull slot") + } + close(siblingRelease) + if err := <-done; err != nil { + t.Fatalf("SyncOnce: %v", err) + } +} diff --git a/internal/mountsync/syncer.go b/internal/mountsync/syncer.go index ebce9742..0df60c53 100644 --- a/internal/mountsync/syncer.go +++ b/internal/mountsync/syncer.go @@ -6622,7 +6622,11 @@ func (s *Syncer) pullRemoteFullGithubTarSeed(ctx context.Context, client githubW var manifest githubCloneManifest var err error s.runFullPullIO(func() { - manifest, err = s.readGithubCloneManifest(ctx) + err = fullPullReadGate.do(ctx, func() error { + var readErr error + manifest, readErr = s.readGithubCloneManifest(ctx) + return readErr + }) }) if err != nil { if exportSnapshotUnsupported(err) { @@ -6679,8 +6683,18 @@ func (s *Syncer) pullRemoteFullGithubTarSeed(ctx context.Context, client githubW return true, fmt.Errorf("github tar seed verification failed: tree listed %d entries, clone manifest expected %d", len(tree), *manifest.FilesExpected) } + // The tar stream is the largest full-pull read: hold one process-wide slot + // from the export request until its body is closed, not just until the + // response headers arrive. Wait for the slot inside runFullPullIO so the + // Syncer mutex is free while sibling scopes own every slot. var tarBody GithubWorkingTreeTar + var releaseTarSlot func() + var slotErr error s.runFullPullIO(func() { + releaseTarSlot, slotErr = fullPullReadGate.acquire(ctx) + if slotErr != nil { + return + } tarBody, err = client.ExportGithubWorkingTreeTar(ctx, s.workspace, GithubWorkingTreeSeedRequest{ Owner: s.githubWorkingTree.Owner, Repo: s.githubWorkingTree.Repo, @@ -6690,6 +6704,10 @@ func (s *Syncer) pullRemoteFullGithubTarSeed(ctx context.Context, client githubW Gzip: false, }) }) + if slotErr != nil { + return true, slotErr + } + defer releaseTarSlot() if err != nil { if exportSnapshotUnsupported(err) { return false, nil @@ -6747,7 +6765,11 @@ func (s *Syncer) pullRemoteFullExport(ctx context.Context, client exportSnapshot var files []RemoteFile var err error s.runFullPullIO(func() { - files, err = client.ExportFiles(exportCtx, s.workspace, s.remoteRoot) + err = fullPullReadGate.do(exportCtx, func() error { + var exportErr error + files, exportErr = client.ExportFiles(exportCtx, s.workspace, s.remoteRoot) + return exportErr + }) }) if err != nil { if exportSnapshotUnsupported(err) { @@ -6977,7 +6999,12 @@ func (s *Syncer) githubWorkingTreeSnapshot(ctx context.Context, prog bootstrapPr maxObservedRevision := "" for { pageStartCursor := cursor - page, err := s.client.ListTree(ctx, s.workspace, s.githubWorkingTree.ContentsRoot, 200, cursor) + var page TreeResponse + err := fullPullReadGate.do(ctx, func() error { + var listErr error + page, listErr = s.client.ListTree(ctx, s.workspace, s.githubWorkingTree.ContentsRoot, 200, cursor) + return listErr + }) if err != nil { return nil, "", err } @@ -7497,7 +7524,11 @@ func (s *Syncer) pullRemoteFullTree(ctx context.Context, conflicted map[string]s if !pageLoaded || loadedDirectory != currentDirectory || loadedCursor != cursor { var err error s.runFullPullIO(func() { - page, err = s.client.ListTree(ctx, s.workspace, currentDirectory, fullTreeTraversalDepth, cursor) + err = fullPullReadGate.do(ctx, func() error { + var listErr error + page, listErr = s.client.ListTree(ctx, s.workspace, currentDirectory, fullTreeTraversalDepth, cursor) + return listErr + }) }) metrics.listCalls++ if err != nil { @@ -8097,7 +8128,12 @@ func (s *Syncer) readBootstrapFilesBulkEach(ctx context.Context, client bulkRead for index, job := range batch { paths[index] = job.RemotePath } - response, err := client.ReadFilesBulk(ctx, s.workspace, paths) + var response BulkReadResponse + err := fullPullReadGate.do(ctx, func() error { + var readErr error + response, readErr = client.ReadFilesBulk(ctx, s.workspace, paths) + return readErr + }) if err != nil { if isBulkReadUnsupported(err) { unsupported = &bulkReadUnsupportedError{remaining: append([]bootstrapReadJob(nil), remaining...)} @@ -8323,7 +8359,12 @@ func (s *Syncer) readBootstrapFilesIndividuallyBatchEach(ctx context.Context, jo resultCh <- bootstrapReadResult{Index: job.Index, RemotePath: job.RemotePath, File: cached} continue } - file, err := s.client.ReadFile(readCtx, s.workspace, job.RemotePath) + var file RemoteFile + err := fullPullReadGate.do(readCtx, func() error { + var readErr error + file, readErr = s.client.ReadFile(readCtx, s.workspace, job.RemotePath) + return readErr + }) if err == nil { s.objectCache.put(file) prog.touch() diff --git a/packages/sdk/typescript/src/relay-cli/command-spec.json b/packages/sdk/typescript/src/relay-cli/command-spec.json index 159a049c..74713484 100644 --- a/packages/sdk/typescript/src/relay-cli/command-spec.json +++ b/packages/sdk/typescript/src/relay-cli/command-spec.json @@ -1076,6 +1076,10 @@ "flags": "--interval-jitter ", "description": "sync interval jitter ratio (0.0-1.0)" }, + { + "flags": "--startup-jitter ", + "description": "maximum random delay before the first sync cycle, so mounts started together (scheduled sandboxes, scoped siblings) do not bootstrap in lockstep (0 disables)" + }, { "flags": "--timeout ", "description": "per-sync timeout"