Merge pull request #361 from AgentWorkforce/fix/bootstrap-empty-revision-360

fix(mount): stop repeated stalled bootstrap checkpoints

Author: miyaontherelay

cmd/relayfile-mount/main.go

@@ -432,7 +432,7 @@ func runSinglePollingMount(rootCtx context.Context, cfg mountConfig) error {
 	log.Printf("%s", mountStartupLogLine(cfg))
 	log.Printf("Mirror started at %s. Sync interval %s +/- %.0f%%. Public state: %s", cfg.localDir, cfg.interval.Round(time.Second), cfg.intervalJitter*100, filepath.Join(cfg.localDir, ".relay", "state.json"))
 
-	run := func(reconcile bool) {
+	run := func(reconcile bool) error {
 		ctx, cancel := context.WithTimeout(rootCtx, cfg.timeout)
 		defer cancel()
 		var err error
@@ -442,19 +442,30 @@ func runSinglePollingMount(rootCtx context.Context, cfg mountConfig) error {
 			err = syncer.SyncOnce(ctx)
 		}
 		if err != nil {
+			var stalled *mountsync.BootstrapStalledError
+			if errors.As(err, &stalled) {
+				// This is an operator-actionable hard stop, not a transient
+				// cycle failure. Returning it terminates this runner (and, for
+				// scoped layouts, cancels sibling runners) instead of letting
+				// the polling ticker retry the same persisted checkpoint forever.
+				return err
+			}
 			if errors.Is(err, context.DeadlineExceeded) {
 				if synced, total, ok := readBootstrapProgress(cfg.localDir); ok {
 					log.Printf("mount bootstrapping: %d/%d files (in progress)", synced, total)
-					return
+					return nil
 				}
 			}
 			log.Printf("mount sync cycle failed: %v", err)
-			return
+			return nil
 		}
 		log.Printf("mount sync cycle completed")
+		return nil
 	}
 
-	run(true)
+	if err := run(true); err != nil {
+		return err
+	}
 	if cfg.once {
 		return nil
 	}
@@ -503,7 +514,9 @@ func runSinglePollingMount(rootCtx context.Context, cfg mountConfig) error {
 				cycle,
 			)
 			if reconcile {
-				run(true)
+				if err := run(true); err != nil {
+					return err
+				}
 			}
 			timer.Reset(jitteredIntervalWithSample(cfg.interval, cfg.intervalJitter, rng.Float64()))
 		}

cmd/relayfile-mount/main_test.go

@@ -2,7 +2,9 @@ package main
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"os"
@@ -415,6 +417,142 @@ func TestInstallCredsFileRefreshToleratesParseFailureWithoutRetry(t *testing.T)
 	}
 }
 
+// TestRunSinglePollingMountStopsOnBootstrapStall proves the typed hard failure
+// leaves the polling runner immediately. main turns this returned error into a
+// nonzero process exit, so the ticker cannot retry the same checkpoint.
+func TestRunSinglePollingMountStopsOnBootstrapStall(t *testing.T) {
+	t.Setenv("RELAYFILE_BOOTSTRAP_STALL_CYCLES", "1")
+	var calls atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		calls.Add(1)
+		http.Error(w, "stuck", http.StatusBadGateway)
+	}))
+	defer server.Close()
+
+	err := runSinglePollingMount(context.Background(), mountConfig{
+		baseURL:          server.URL,
+		token:            "test-token",
+		workspaceID:      "ws_bootstrap_stall",
+		remotePath:       "/",
+		localDir:         t.TempDir(),
+		stateDir:         t.TempDir(),
+		mountKind:        mountsync.MountKindDaemon,
+		syncMode:         syncModeMirror,
+		interval:         time.Hour,
+		timeout:          time.Second,
+		websocketEnabled: false,
+	})
+	var stalled *mountsync.BootstrapStalledError
+	if !errors.As(err, &stalled) {
+		t.Fatalf("expected bootstrap stall to escape polling runner, got %v", err)
+	}
+	if got := calls.Load(); got == 0 {
+		t.Fatal("expected initial full-tree request before runner exited")
+	}
+}
+
+// TestRunSinglePollingMountKeepsNormalCycleFailuresNonFatal verifies that a
+// normal cloud error keeps its historical per-cycle retry behavior rather
+// than terminating the mount runner.
+func TestRunSinglePollingMountKeepsNormalCycleFailuresNonFatal(t *testing.T) {
+	t.Setenv("RELAYFILE_BOOTSTRAP_STALL_CYCLES", "2")
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "transient", http.StatusBadGateway)
+	}))
+	defer server.Close()
+
+	err := runSinglePollingMount(context.Background(), mountConfig{
+		baseURL:          server.URL,
+		token:            "test-token",
+		workspaceID:      "ws_normal_retry",
+		remotePath:       "/",
+		localDir:         t.TempDir(),
+		stateDir:         t.TempDir(),
+		mountKind:        mountsync.MountKindDaemon,
+		syncMode:         syncModeMirror,
+		interval:         time.Hour,
+		timeout:          time.Second,
+		websocketEnabled: false,
+		once:             true,
+	})
+	if err != nil {
+		t.Fatalf("normal cycle failure must remain nonfatal to the runner, got %v", err)
+	}
+}
+
+// TestRunSinglePollingMountStopsOnTimerBootstrapStall exercises the polling
+// timer path, not just the initial cycle. The first page commits a partial
+// checkpoint and its next-page error remains nonfatal; the following timer
+// reconcile fails at that unchanged cursor and terminates the runner.
+func TestRunSinglePollingMountStopsOnTimerBootstrapStall(t *testing.T) {
+	t.Setenv("RELAYFILE_BOOTSTRAP_STALL_CYCLES", "1")
+	var rootTreeCalls atomic.Int32
+	var cursorTreeCalls atomic.Int32
+	var readCalls atomic.Int32
+	entries := make([]mountsync.TreeEntry, 0, 10)
+	for i := 0; i < 10; i++ {
+		entries = append(entries, mountsync.TreeEntry{Path: fmt.Sprintf("/f/%05d.txt", i), Type: "file"})
+	}
+	nextCursor := entries[len(entries)-1].Path
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch {
+		case strings.Contains(r.URL.Path, "/fs/tree"):
+			if r.URL.Query().Get("cursor") != "" {
+				cursorTreeCalls.Add(1)
+				// A 400 is intentionally not retried by HTTPClient, so exactly
+				// two calls prove the initial nonfatal cycle plus one timer cycle.
+				http.Error(w, "stuck cursor", http.StatusBadRequest)
+				return
+			}
+			rootTreeCalls.Add(1)
+			w.Header().Set("Content-Type", "application/json")
+			_ = json.NewEncoder(w).Encode(mountsync.TreeResponse{Entries: entries, NextCursor: &nextCursor})
+		case strings.Contains(r.URL.Path, "/fs/file"):
+			readCalls.Add(1)
+			w.Header().Set("Content-Type", "application/json")
+			_ = json.NewEncoder(w).Encode(mountsync.RemoteFile{
+				Path:        r.URL.Query().Get("path"),
+				ContentType: "text/plain",
+				Content:     "content",
+			})
+		default:
+			http.NotFound(w, r)
+		}
+	}))
+	defer server.Close()
+
+	started := time.Now()
+	err := runSinglePollingMount(context.Background(), mountConfig{
+		baseURL:          server.URL,
+		token:            "test-token",
+		workspaceID:      "ws_timer_bootstrap_stall",
+		remotePath:       "/",
+		localDir:         t.TempDir(),
+		stateDir:         t.TempDir(),
+		mountKind:        mountsync.MountKindDaemon,
+		syncMode:         syncModeMirror,
+		interval:         time.Millisecond, // enforced to the 5s poll floor
+		timeout:          time.Second,
+		websocketEnabled: false,
+	})
+	var stalled *mountsync.BootstrapStalledError
+	if !errors.As(err, &stalled) {
+		t.Fatalf("expected timer-path bootstrap stall to escape runner, got %v", err)
+	}
+	if got := rootTreeCalls.Load(); got != 1 {
+		t.Fatalf("root tree calls = %d, want one initial partial traversal", got)
+	}
+	if got := readCalls.Load(); got != int32(len(entries)) {
+		t.Fatalf("ReadFile calls = %d, want %d first-page files", got, len(entries))
+	}
+	if got := cursorTreeCalls.Load(); got != 2 {
+		t.Fatalf("cursor tree calls = %d, want initial nonfatal + timer hard-stop", got)
+	}
+	if elapsed := time.Since(started); elapsed < minMountPollInterval {
+		t.Fatalf("runner returned before a timer cycle (%s < %s)", elapsed, minMountPollInterval)
+	}
+}
+
 func TestNormalizeRemotePathsDedupesRepeatedFlagValues(t *testing.T) {
 	got := normalizeRemotePaths(
 		[]string{"/github/repos/acme/cloud", "github/repos/acme/cloud/", "/slack/channels/proj-cloud"},

internal/mountsync/bootstrap_test.go

@@ -3,6 +3,7 @@ package mountsync
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -365,6 +366,173 @@ func TestBootstrapResumesFromPersistedCursor(t *testing.T) {
 	}
 }
 
+// TestBootstrapStallCycleGuardPersistsAndFailsHard reproduces the
+// non-converging unversioned partial-bootstrap shape: the first page is
+// persisted, then every resumed cycle fails at the same next-page cursor.
+// The empty revision is non-destructive and not the reason completion fails;
+// the unchanged checkpoint is. The guard turns the otherwise infinite
+// "non-empty state without completed bootstrap" loop into a hard failure,
+// even when every attempt is a fresh Syncer process.
+func TestBootstrapStallCycleGuardPersistsAndFailsHard(t *testing.T) {
+	client := newBootstrapClient(30, 10)
+	for path, file := range client.files {
+		file.Revision = ""
+		client.files[path] = file
+	}
+	client.listTreeFailAt = 2
+	client.listTreeFailErr = &HTTPError{StatusCode: 502, Code: "bad_gateway", Message: "stuck"}
+	localDir := t.TempDir()
+	opts := SyncerOptions{RootCtx: context.Background(), BootstrapStallCycles: 2}
+
+	s := newBootstrapSyncer(t, client, localDir, opts)
+	err := s.Reconcile(context.Background())
+	if err == nil {
+		t.Fatalf("expected partial bootstrap cycle to return its cloud error")
+	}
+	var stalled *BootstrapStalledError
+	if errors.As(err, &stalled) {
+		t.Fatalf("checkpoint-advancing cycle must stay below the hard limit: %v", err)
+	}
+	st := loadPersistedState(t, localDir)
+	if st.BootstrapStallCycles != 0 || st.BootstrapCursor == "" || st.BootstrapFilesSynced == 0 {
+		t.Fatalf("expected persisted partial checkpoint with reset stall count, state=%#v", st)
+	}
+	if got := countLocalFiles(t, localDir); got != 10 {
+		t.Fatalf("expected first page to stay materialized, got %d files", got)
+	}
+
+	// A fresh Syncer resumes at the same failing cursor. This is the first
+	// checkpoint-stable stalled cycle, so it increments but stays retryable.
+	s = newBootstrapSyncer(t, client, localDir, opts)
+	err = s.Reconcile(context.Background())
+	if err == nil || errors.As(err, &stalled) {
+		t.Fatalf("first checkpoint-stable cycle must return the cloud error, got %v", err)
+	}
+	st = loadPersistedState(t, localDir)
+	if st.BootstrapStallCycles != 1 {
+		t.Fatalf("BootstrapStallCycles after first unchanged cycle = %d, want 1", st.BootstrapStallCycles)
+	}
+
+	// The next fresh Syncer sees the same persisted partial checkpoint and
+	// reaches the configured limit, returning a typed, status-visible failure.
+	s = newBootstrapSyncer(t, client, localDir, opts)
+	err = s.Reconcile(context.Background())
+	if !errors.As(err, &stalled) {
+		t.Fatalf("expected BootstrapStalledError at the hard limit, got %v", err)
+	}
+	if stalled.Cycles != 2 || stalled.Limit != 2 {
+		t.Fatalf("BootstrapStalledError = %#v, want cycles=2 limit=2", stalled)
+	}
+	st = loadPersistedState(t, localDir)
+	if st.BootstrapComplete {
+		t.Fatalf("hard stall must not force BootstrapComplete")
+	}
+	if st.BootstrapStallCycles != 2 {
+		t.Fatalf("persisted BootstrapStallCycles = %d, want 2", st.BootstrapStallCycles)
+	}
+	if st.LastError == nil || st.LastError.Kind != "bootstrap_stalled" || st.LastError.Code != "bootstrap_stall_cycle_limit" {
+		t.Fatalf("expected structured persisted bootstrap stall error, got %#v", st.LastError)
+	}
+}
+
+func TestBootstrapStallCycleGuardIgnoresCanceledContext(t *testing.T) {
+	s := newBootstrapSyncer(t, newBootstrapClient(1, 1), t.TempDir(), SyncerOptions{
+		RootCtx:              context.Background(),
+		BootstrapStallCycles: 2,
+	})
+	s.state.BootstrapCursor = "resume-cursor"
+	s.state.BootstrapStallCycles = 1
+
+	wrappedCanceled := fmt.Errorf("mount shutting down: %w", context.Canceled)
+	if err := s.recordBootstrapCycle("resume-cursor", nil, wrappedCanceled); err != nil {
+		t.Fatalf("canceled context must not trip the stall guard: %v", err)
+	}
+	if got := s.state.BootstrapStallCycles; got != 1 {
+		t.Fatalf("canceled context changed stall count to %d, want 1", got)
+	}
+
+	// DeadlineExceeded is a failed retry at the same checkpoint, so it still
+	// consumes the remaining attempt and produces the typed hard stop.
+	err := s.recordBootstrapCycle("resume-cursor", nil, context.DeadlineExceeded)
+	var stalled *BootstrapStalledError
+	if !errors.As(err, &stalled) {
+		t.Fatalf("deadline exceeded must count toward the stall limit, got %v", err)
+	}
+	if stalled.Cycles != 2 || stalled.Limit != 2 {
+		t.Fatalf("BootstrapStalledError = %#v, want cycles=2 limit=2", stalled)
+	}
+}
+
+// TestBootstrapStallCycleGuardResetsOnCheckpointAdvanceAndCompletion proves
+// that the counter reflects lack of traversal progress, not merely errors.
+func TestBootstrapStallCycleGuardResetsOnCheckpointAdvanceAndCompletion(t *testing.T) {
+	client := newBootstrapClient(30, 10)
+	client.listTreeFailAt = 2
+	client.listTreeFailErr = &HTTPError{StatusCode: 502, Code: "bad_gateway", Message: "page failure"}
+	localDir := t.TempDir()
+	s := newBootstrapSyncer(t, client, localDir, SyncerOptions{
+		RootCtx:              context.Background(),
+		BootstrapStallCycles: 3,
+	})
+
+	// Page one advances the cursor before page two fails, so the count resets.
+	if err := s.Reconcile(context.Background()); err == nil {
+		t.Fatalf("expected first bootstrap attempt to fail on page 2")
+	}
+	if st := loadPersistedState(t, localDir); st.BootstrapStallCycles != 0 || st.BootstrapCursor == "" {
+		t.Fatalf("cursor advance must reset stall count; state=%#v", st)
+	}
+
+	// Fail at the same persisted cursor once: count increments.
+	if err := s.Reconcile(context.Background()); err == nil {
+		t.Fatalf("expected unchanged checkpoint attempt to fail")
+	}
+	if st := loadPersistedState(t, localDir); st.BootstrapStallCycles != 1 {
+		t.Fatalf("unchanged checkpoint count = %d, want 1", st.BootstrapStallCycles)
+	}
+
+	// Let page two advance, then fail on page three. The new cursor resets the
+	// persisted count despite the returned cloud error.
+	client.mu.Lock()
+	client.listTreeFailAt = 3
+	client.mu.Unlock()
+	if err := s.Reconcile(context.Background()); err == nil {
+		t.Fatalf("expected bootstrap attempt to fail on page 3")
+	}
+	if st := loadPersistedState(t, localDir); st.BootstrapStallCycles != 0 {
+		t.Fatalf("advanced checkpoint must reset stall count, got %d", st.BootstrapStallCycles)
+	}
+
+	client.mu.Lock()
+	client.listTreeFailAt = 0
+	client.mu.Unlock()
+	if err := s.Reconcile(context.Background()); err != nil {
+		t.Fatalf("completion after checkpoint advances: %v", err)
+	}
+	st := loadPersistedState(t, localDir)
+	if !st.BootstrapComplete || st.BootstrapStallCycles != 0 {
+		t.Fatalf("completion must clear checkpoint and stall count, state=%#v", st)
+	}
+}
+
+func TestBootstrapStallCycleLimitUsesOptionThenEnvThenDefault(t *testing.T) {
+	newSyncer := func(t *testing.T, opts SyncerOptions) *Syncer {
+		t.Helper()
+		return newBootstrapSyncer(t, newBootstrapClient(1, 1), t.TempDir(), opts)
+	}
+	t.Setenv("RELAYFILE_BOOTSTRAP_STALL_CYCLES", "7")
+	if got := newSyncer(t, SyncerOptions{}).bootstrapStallCycles; got != 7 {
+		t.Fatalf("env bootstrap stall limit = %d, want 7", got)
+	}
+	if got := newSyncer(t, SyncerOptions{BootstrapStallCycles: 3}).bootstrapStallCycles; got != 3 {
+		t.Fatalf("explicit bootstrap stall limit = %d, want 3", got)
+	}
+	t.Setenv("RELAYFILE_BOOTSTRAP_STALL_CYCLES", "not-a-number")
+	if got := newSyncer(t, SyncerOptions{}).bootstrapStallCycles; got != defaultBootstrapStallCycles {
+		t.Fatalf("invalid env bootstrap stall limit = %d, want default %d", got, defaultBootstrapStallCycles)
+	}
+}
+
 // TestBootstrapCompleteGatesFastPath: a hand-seeded state with Files +
 // LastEventAt but BootstrapComplete=false MUST NOT short-circuit; the
 // full pull runs and the mirror converges (the rw_517d60b6 repro).