From 066d826dffc9276d9f8d574fa90b107eba82c5ef Mon Sep 17 00:00:00 2001 From: agentrelaybot Date: Tue, 29 Sep 2026 23:57:34 -0700 Subject: [PATCH 1/6] fix(mount): startup splay and process-wide full-pull read cap Mounts started in the same instant (scheduled sandboxes at cron boundaries, scoped siblings in one process) ran their first, possibly full-tree, reconcile immediately and in lockstep. Scoped layouts also run one Syncer per remote path, each with its own 4-worker bootstrap pool, so N scopes issued 4N concurrent reads against the single-threaded workspace Durable Object. - relayfile-mount: --startup-jitter / RELAYFILE_MOUNT_STARTUP_JITTER (default 5s, max 5m, 0 disables) delays the first cycle by a uniform random amount; cancellation during the splay stops cleanly (and fails --once, which has not bootstrapped). - mountsync: a process-wide gate caps in-flight full-pull reads (tree pages, bulk reads, bootstrap point reads, export snapshots) at 4 across all Syncers (RELAYFILE_FULL_PULL_READ_CONCURRENCY, max 16). Waiting honours ctx, so a timed-out cycle yields exactly as before. Co-Authored-By: Claude Opus 5.5 (1M context) Session-Id: 04a50716-3a69-4bf5-b6b2-7696cd1021cd --- cmd/relayfile-mount/main.go | 59 ++++++++++++++ cmd/relayfile-mount/startup_splay_test.go | 93 +++++++++++++++++++++++ internal/mountsync/fullpull_gate.go | 64 ++++++++++++++++ internal/mountsync/fullpull_gate_test.go | 84 ++++++++++++++++++++ internal/mountsync/syncer.go | 33 ++++++-- 5 files changed, 328 insertions(+), 5 deletions(-) create mode 100644 cmd/relayfile-mount/startup_splay_test.go create mode 100644 internal/mountsync/fullpull_gate.go create mode 100644 internal/mountsync/fullpull_gate_test.go diff --git a/cmd/relayfile-mount/main.go b/cmd/relayfile-mount/main.go index ea7a8545..90a5d7eb 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,6 +732,13 @@ func runSinglePollingMount(rootCtx context.Context, cfg mountConfig) error { // "just finished as part of this attempt." priorBootstrapComplete := bootstrapAlreadyComplete(cfg.localDir) + if err := waitStartupSplay(rootCtx, startupSplayDelay(cfg.startupJitter, rand.Float64())); 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 err := run(true); err != nil { return err } @@ -1532,6 +1544,53 @@ 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 +) + +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 waitStartupSplay(ctx context.Context, delay time.Duration) error { + if delay <= 0 { + return 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 nil + case <-ctx.Done(): + return ctx.Err() + } +} + func clampJitterRatio(value float64) float64 { if value < 0 { return 0 diff --git a/cmd/relayfile-mount/startup_splay_test.go b/cmd/relayfile-mount/startup_splay_test.go new file mode 100644 index 00000000..a43a1f58 --- /dev/null +++ b/cmd/relayfile-mount/startup_splay_test.go @@ -0,0 +1,93 @@ +package main + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "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, + } + ctx, cancel := context.WithTimeout(context.Background(), 150*time.Millisecond) + defer cancel() + // rand.Float64() is almost never below 150ms/5m; retry the rare draw that + // is, so the assertion never flakes. + for attempt := 0; attempt < 3; attempt++ { + requests.Store(0) + err := runSinglePollingMount(ctx, cfg) + if err == nil { + ctx, cancel = context.WithTimeout(context.Background(), 150*time.Millisecond) + defer cancel() + continue + } + if got := requests.Load(); got != 0 { + t.Fatalf("mount made %d request(s) during the startup splay, want 0", got) + } + return + } + t.Fatal("startup splay never delayed the first cycle") +} + +func TestWaitStartupSplayZeroDelayRunsImmediately(t *testing.T) { + start := time.Now() + if err := waitStartupSplay(context.Background(), 0); err != nil { + t.Fatalf("zero splay: %v", err) + } + if elapsed := time.Since(start); elapsed > 50*time.Millisecond { + t.Fatalf("zero splay waited %s", elapsed) + } +} diff --git a/internal/mountsync/fullpull_gate.go b/internal/mountsync/fullpull_gate.go new file mode 100644 index 00000000..c4e1338f --- /dev/null +++ b/internal/mountsync/fullpull_gate.go @@ -0,0 +1,64 @@ +package mountsync + +import ( + "context" + "os" + "strconv" + "strings" +) + +// defaultFullPullReadConcurrency caps in-flight full-tree remote reads (tree +// pages, bulk reads, bootstrap point reads, export snapshots) 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 { + select { + case g.slots <- struct{}{}: + case <-ctx.Done(): + return ctx.Err() + } + defer func() { <-g.slots }() + return fn() +} + +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..3fab9b0b --- /dev/null +++ b/internal/mountsync/fullpull_gate_test.go @@ -0,0 +1,84 @@ +package mountsync + +import ( + "context" + "sync" + "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") + 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) + } + } +} diff --git a/internal/mountsync/syncer.go b/internal/mountsync/syncer.go index ebce9742..355e0fa8 100644 --- a/internal/mountsync/syncer.go +++ b/internal/mountsync/syncer.go @@ -6747,7 +6747,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 +6981,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 +7506,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 +8110,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 +8341,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() From f0926f7d27eb61096214660bd3e9098edfb7a8b7 Mon Sep 17 00:00:00 2001 From: agentrelaybot Date: Wed, 30 Sep 2026 05:50:15 -0700 Subject: [PATCH 2/6] fix(mountsync): keep GitHub tar seeds inside the full-pull read cap The GitHub working-tree seed read its clone manifest outside fullPullReadGate and released nothing for its tar export, whose body is streamed after the request returns. N GitHub scopes seeding together therefore ran N unbounded manifest reads and tar streams on top of the four gated tree/bulk reads. - Gate the clone-manifest read. - Hold one slot from the tar export request until its body is closed (readGate.acquire returns an idempotent release). - Tests install their own gate (withFullPullReadGate) so the cap assertion no longer depends on RELAYFILE_FULL_PULL_READ_CONCURRENCY in the environment that initialised the package gate. Co-Authored-By: Claude Opus 5.5 (1M context) Session-Id: cfcb450a-cb3d-44b8-b164-ee16099b74f9 --- internal/mountsync/fullpull_gate.go | 22 +++- internal/mountsync/fullpull_gate_test.go | 130 +++++++++++++++++++++++ internal/mountsync/syncer.go | 14 ++- 3 files changed, 161 insertions(+), 5 deletions(-) diff --git a/internal/mountsync/fullpull_gate.go b/internal/mountsync/fullpull_gate.go index c4e1338f..51e62dd5 100644 --- a/internal/mountsync/fullpull_gate.go +++ b/internal/mountsync/fullpull_gate.go @@ -5,10 +5,12 @@ import ( "os" "strconv" "strings" + "sync" ) // defaultFullPullReadConcurrency caps in-flight full-tree remote reads (tree -// pages, bulk reads, bootstrap point reads, export snapshots) across EVERY +// 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 @@ -39,13 +41,25 @@ func newReadGate(limit int) *readGate { // 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 ctx.Err() + return nil, ctx.Err() } - defer func() { <-g.slots }() - return fn() + var once sync.Once + return func() { once.Do(func() { <-g.slots }) }, nil } func fullPullReadConcurrency() int { diff --git a/internal/mountsync/fullpull_gate_test.go b/internal/mountsync/fullpull_gate_test.go index 3fab9b0b..e6c35a8b 100644 --- a/internal/mountsync/fullpull_gate_test.go +++ b/internal/mountsync/fullpull_gate_test.go @@ -2,7 +2,9 @@ package mountsync import ( "context" + "io" "sync" + "sync/atomic" "testing" "time" ) @@ -13,6 +15,7 @@ import ( // 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 @@ -82,3 +85,130 @@ func TestFullPullReadConcurrencyEnv(t *testing.T) { } } } + +// 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") + } +} diff --git a/internal/mountsync/syncer.go b/internal/mountsync/syncer.go index 355e0fa8..633044bf 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,6 +6683,14 @@ 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. + releaseTarSlot, err := fullPullReadGate.acquire(ctx) + if err != nil { + return true, err + } + defer releaseTarSlot() var tarBody GithubWorkingTreeTar s.runFullPullIO(func() { tarBody, err = client.ExportGithubWorkingTreeTar(ctx, s.workspace, GithubWorkingTreeSeedRequest{ From 24067e10c517bbbd2665b014d70cea804d196abe Mon Sep 17 00:00:00 2001 From: agentrelaybot Date: Wed, 30 Sep 2026 05:50:24 -0700 Subject: [PATCH 3/6] fix(relayfile-mount): service flush requests during the startup splay A SIGUSR1 flush that arrived while the daemon was still waiting out its startup splay (up to 5m) stayed queued until the splay ended, so `--notify-flush` gave up after notifyFlushWait (2x the cycle timeout) without an ack. A flush now ends the splay and is serviced immediately through the same handler as the main loop (serviceFlushRequest). The splay test pins its sample (startupSplaySample) instead of retrying a random draw, so it no longer depends on the cycle's outcome. Co-Authored-By: Claude Opus 5.5 (1M context) Session-Id: cfcb450a-cb3d-44b8-b164-ee16099b74f9 --- cmd/relayfile-mount/main.go | 42 +++++++---- cmd/relayfile-mount/notify_flush.go | 19 +++++ cmd/relayfile-mount/startup_splay_test.go | 92 +++++++++++++++++++---- 3 files changed, 121 insertions(+), 32 deletions(-) diff --git a/cmd/relayfile-mount/main.go b/cmd/relayfile-mount/main.go index 90a5d7eb..9417ff34 100644 --- a/cmd/relayfile-mount/main.go +++ b/cmd/relayfile-mount/main.go @@ -732,14 +732,23 @@ func runSinglePollingMount(rootCtx context.Context, cfg mountConfig) error { // "just finished as part of this attempt." priorBootstrapComplete := bootstrapAlreadyComplete(cfg.localDir) - if err := waitStartupSplay(rootCtx, startupSplayDelay(cfg.startupJitter, rand.Float64())); 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 err := run(true); err != 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 { @@ -803,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) { @@ -1554,6 +1555,10 @@ const ( 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 @@ -1576,18 +1581,23 @@ func startupSplayDelay(max time.Duration, sample float64) time.Duration { return time.Duration(sample * float64(max)) } -func waitStartupSplay(ctx context.Context, delay time.Duration) error { +// 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 ctx.Err() + 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 nil + return false, nil + case <-flushReq: + log.Printf("mount startup splay: flush requested; starting first sync now") + return true, nil case <-ctx.Done(): - return ctx.Err() + return false, ctx.Err() } } 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 index a43a1f58..73e954ee 100644 --- a/cmd/relayfile-mount/startup_splay_test.go +++ b/cmd/relayfile-mount/startup_splay_test.go @@ -9,6 +9,7 @@ import ( "testing" "time" + "github.com/agentworkforce/relayfile/internal/mountlease" "github.com/agentworkforce/relayfile/internal/mountsync" ) @@ -62,32 +63,91 @@ func TestRunSinglePollingMountWaitsStartupSplayBeforeFirstRequest(t *testing.T) 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() - // rand.Float64() is almost never below 150ms/5m; retry the rare draw that - // is, so the assertion never flakes. - for attempt := 0; attempt < 3; attempt++ { - requests.Store(0) - err := runSinglePollingMount(ctx, cfg) - if err == nil { - ctx, cancel = context.WithTimeout(context.Background(), 150*time.Millisecond) - defer cancel() - continue - } - if got := requests.Load(); got != 0 { - t.Fatalf("mount made %d request(s) during the startup splay, want 0", got) - } - return + err := runSinglePollingMount(ctx, cfg) + if err == nil { + t.Fatal("one-shot mount returned success while its startup splay outlasted the deadline") } - t.Fatal("startup splay never delayed the first cycle") + 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); err != nil { + 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)) +} From 389705d4659480cad70de08417117070f3d7afcc Mon Sep 17 00:00:00 2001 From: agentrelaybot Date: Wed, 30 Sep 2026 05:50:31 -0700 Subject: [PATCH 4/6] fix(cli): splay the first sync of `relayfile mount` too The public `relayfile mount` command runs its own loop (runMountLoopWithAuthLock), so only standalone relayfile-mount waited out the startup splay; simultaneous public mounts still bootstrapped in lockstep. Add --startup-jitter / RELAYFILE_MOUNT_STARTUP_JITTER (default 5s, max 5m) and wait before the first cycle. Each scoped runner calls the loop separately and draws its own delay; cancellation keeps the existing once-mode error and daemon-mode clean stop. Co-Authored-By: Claude Opus 5.5 (1M context) Session-Id: cfcb450a-cb3d-44b8-b164-ee16099b74f9 --- cmd/relayfile-cli/main.go | 60 ++++++++++++++++++- cmd/relayfile-cli/startup_splay_test.go | 77 +++++++++++++++++++++++++ 2 files changed, 136 insertions(+), 1 deletion(-) create mode 100644 cmd/relayfile-cli/startup_splay_test.go 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) + } +} From eb02a28c7f50c2258245f5016b3e7f79df8521ef Mon Sep 17 00:00:00 2001 From: agentrelaybot Date: Wed, 30 Sep 2026 06:01:37 -0700 Subject: [PATCH 5/6] fix(mountsync): wait for the tar-seed slot without holding the Syncer mutex The GitHub tar seed acquired its full-pull slot before runFullPullIO released s.mu, so while sibling scopes owned every slot this Syncer's local writeback, outbox and watcher handling were blocked for the rest of their streams. Acquire inside runFullPullIO like every other gated read, and do not record a gate wait cancellation as a cloud failure (Bugbot). Co-Authored-By: Claude Opus 5.5 (1M context) Session-Id: cfcb450a-cb3d-44b8-b164-ee16099b74f9 --- internal/mountsync/fullpull_gate_test.go | 89 ++++++++++++++++++++++++ internal/mountsync/syncer.go | 18 +++-- 2 files changed, 101 insertions(+), 6 deletions(-) diff --git a/internal/mountsync/fullpull_gate_test.go b/internal/mountsync/fullpull_gate_test.go index e6c35a8b..f6ea5a61 100644 --- a/internal/mountsync/fullpull_gate_test.go +++ b/internal/mountsync/fullpull_gate_test.go @@ -212,3 +212,92 @@ func TestGithubTarSeedHoldsFullPullSlotUntilBodyClosed(t *testing.T) { 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 633044bf..0df60c53 100644 --- a/internal/mountsync/syncer.go +++ b/internal/mountsync/syncer.go @@ -6685,14 +6685,16 @@ func (s *Syncer) pullRemoteFullGithubTarSeed(ctx context.Context, client githubW // 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. - releaseTarSlot, err := fullPullReadGate.acquire(ctx) - if err != nil { - return true, err - } - defer releaseTarSlot() + // 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, @@ -6702,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 From 76c8e2b975d1bccf8afe456fc89f2c038cc3f05d Mon Sep 17 00:00:00 2001 From: agentrelaybot Date: Wed, 30 Sep 2026 06:16:08 -0700 Subject: [PATCH 6/6] fix(cli): declare --startup-jitter in the mount command spec TestOptionsMatchSourceFlagSets requires every flag runMount registers to be declared in the command table; regenerate the SDK command-spec snapshot with gen-command-spec.mjs. Co-Authored-By: Claude Opus 5.5 (1M context) Session-Id: cfcb450a-cb3d-44b8-b164-ee16099b74f9 --- cmd/relayfile-cli/commandspec.go | 1 + packages/sdk/typescript/src/relay-cli/command-spec.json | 4 ++++ 2 files changed, 5 insertions(+) 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/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"