Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 18 additions & 5 deletions cmd/relayfile-mount/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
}
Expand Down Expand Up @@ -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()))
}
Expand Down
138 changes: 138 additions & 0 deletions cmd/relayfile-mount/main_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,9 @@ package main

import (
"context"
"encoding/json"
"errors"
"fmt"
"net/http"
"net/http/httptest"
"os"
Expand Down Expand Up @@ -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"},
Expand Down
168 changes: 168 additions & 0 deletions internal/mountsync/bootstrap_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package mountsync
import (
"context"
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
Expand Down Expand Up @@ -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).
Expand Down
Loading
Loading