wehub-resource-sync
f99010fae1
CI / lint (push) Failing after 1s
CI / frontend (push) Failing after 1s
CI / scripts (push) Failing after 1s
CI / Go Test (ubuntu-latest) (push) Failing after 0s
CI / frontend-node-25 (push) Failing after 1s
CI / docs (push) Failing after 0s
CI / coverage (push) Failing after 0s
CI / e2e (push) Failing after 0s
Docker / build-and-push (push) Failing after 1s
CI / integration (push) Failing after 4m43s
CI / Go Test (windows-latest) (push) Has been cancelled
CI / Desktop Unit Tests (Windows) (push) Has been cancelled
Desktop Artifacts / Desktop Build (Linux (arm64)) (push) Has been cancelled
Desktop Artifacts / Desktop Build (Linux) (push) Has been cancelled
Desktop Artifacts / Desktop Build (Windows) (push) Has been cancelled
Desktop Artifacts (macOS) / Desktop Build (macOS (aarch64)) (push) Has been cancelled
Desktop Artifacts (macOS) / Desktop Build (macOS (x86_64)) (push) Has been cancelled
2605 lines
85 KiB
Go
2605 lines
85 KiB
Go
package remotesync
|
|
|
|
import (
|
|
"archive/tar"
|
|
"bytes"
|
|
"compress/gzip"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"os"
|
|
"path/filepath"
|
|
"runtime"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
stdsync "sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
"go.kenn.io/agentsview/internal/db"
|
|
"go.kenn.io/agentsview/internal/parser"
|
|
syncpkg "go.kenn.io/agentsview/internal/sync"
|
|
"go.kenn.io/agentsview/internal/testjsonl"
|
|
)
|
|
|
|
func TestHTTPSyncDownloadsArchiveAndImports(t *testing.T) {
|
|
archive := buildHTTPTestTar(t, map[string]string{
|
|
"home/wes/.claude/projects/test-project/session.jsonl": testjsonl.NewSessionBuilder().
|
|
AddClaudeUser("2024-01-01T00:00:00Z", "http remote").
|
|
String(),
|
|
})
|
|
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
assert.Equal(t, "Bearer remote-token", r.Header.Get("Authorization"))
|
|
switch r.URL.Path {
|
|
case "/api/v1/remote-sync/targets":
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_, _ = w.Write([]byte(`{"dirs":{"claude":["/home/wes/.claude/projects"]}}`))
|
|
case "/api/v1/remote-sync/archive":
|
|
w.Header().Set("Content-Type", "application/x-tar")
|
|
_, _ = w.Write(archive)
|
|
default:
|
|
http.NotFound(w, r)
|
|
}
|
|
}))
|
|
t.Cleanup(ts.Close)
|
|
|
|
database, err := db.Open(filepath.Join(t.TempDir(), "test.db"))
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { require.NoError(t, database.Close()) })
|
|
|
|
stats, err := HTTPSync{
|
|
Host: "devbox",
|
|
URL: ts.URL,
|
|
Token: "remote-token",
|
|
DB: database,
|
|
}.Run(context.Background())
|
|
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 1, stats.SessionsSynced)
|
|
}
|
|
|
|
func TestHTTPSyncReportsDownloadAndImportProgress(t *testing.T) {
|
|
archive := buildHTTPTestTar(t, map[string]string{
|
|
"home/wes/.claude/projects/test-project/session.jsonl": testjsonl.NewSessionBuilder().
|
|
AddClaudeUser("2024-01-01T00:00:00Z", "http remote progress").
|
|
String(),
|
|
})
|
|
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
switch r.URL.Path {
|
|
case "/api/v1/remote-sync/targets":
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_, _ = w.Write([]byte(`{"dirs":{"claude":["/home/wes/.claude/projects"]}}`))
|
|
case "/api/v1/remote-sync/archive":
|
|
w.Header().Set("Content-Type", "application/x-tar")
|
|
w.Header().Set("Content-Length", strconv.Itoa(len(archive)))
|
|
_, _ = w.Write(archive)
|
|
default:
|
|
http.NotFound(w, r)
|
|
}
|
|
}))
|
|
t.Cleanup(ts.Close)
|
|
|
|
database, err := db.Open(filepath.Join(t.TempDir(), "test.db"))
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { require.NoError(t, database.Close()) })
|
|
var progress []syncpkg.Progress
|
|
|
|
stats, err := HTTPSync{
|
|
Host: "devbox",
|
|
URL: ts.URL,
|
|
Token: "remote-token",
|
|
DB: database,
|
|
Progress: func(p syncpkg.Progress) {
|
|
progress = append(progress, p)
|
|
},
|
|
}.Run(context.Background())
|
|
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 1, stats.SessionsSynced)
|
|
assert.Contains(t, progressDetails(progress),
|
|
"Resolving agent directories on devbox")
|
|
assert.Contains(t, progressDetails(progress),
|
|
"Downloading session archive from devbox")
|
|
assert.Contains(t, progressDetails(progress),
|
|
"Extracting session archive from devbox")
|
|
assert.Contains(t, progressDetails(progress),
|
|
"Processing sessions from devbox")
|
|
require.NotEmpty(t, progress, "expected progress events")
|
|
assert.Equal(t, int64(len(archive)), maxBytesDone(progress))
|
|
assert.Equal(t, int64(len(archive)), maxBytesTotal(progress))
|
|
}
|
|
|
|
func TestHTTPSyncReportsCompressedTransferThenExtraction(t *testing.T) {
|
|
archive := buildHTTPTestTar(t, map[string]string{
|
|
"home/user/.claude/projects/test/session.jsonl": "compressed archive",
|
|
})
|
|
var compressed bytes.Buffer
|
|
gz := gzip.NewWriter(&compressed)
|
|
_, err := gz.Write(archive)
|
|
require.NoError(t, err)
|
|
require.NoError(t, gz.Close())
|
|
|
|
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
assert.Equal(t, "/api/v1/remote-sync/archive", r.URL.Path)
|
|
w.Header().Set("Content-Encoding", "gzip")
|
|
w.Header().Set("Content-Length", strconv.Itoa(compressed.Len()))
|
|
_, _ = w.Write(compressed.Bytes())
|
|
}))
|
|
t.Cleanup(ts.Close)
|
|
|
|
var progress []syncpkg.Progress
|
|
hs := HTTPSync{
|
|
Host: "devbox",
|
|
URL: ts.URL,
|
|
Progress: func(p syncpkg.Progress) {
|
|
progress = append(progress, p)
|
|
},
|
|
}
|
|
mirrorRoot := filepath.Join(t.TempDir(), "remote-mirrors", "devbox")
|
|
err = hs.downloadIntoMirror(
|
|
context.Background(), ts.Client(), TargetSet{}, []string{"session.jsonl"},
|
|
false, mirrorRoot,
|
|
)
|
|
require.NoError(t, err)
|
|
|
|
downloadLabel := "Downloading 1 changed files from devbox"
|
|
extractLabel := "Extracting 1 changed files from devbox"
|
|
assert.Equal(t, []string{downloadLabel, extractLabel},
|
|
uniqueProgressDetails(progress))
|
|
require.NotEmpty(t, progress)
|
|
assert.Equal(t, int64(compressed.Len()), maxBytesDone(progress))
|
|
assert.Equal(t, int64(compressed.Len()), maxBytesTotal(progress))
|
|
}
|
|
|
|
func TestPrepareHTTPSyncReportsManifestComparisonBeforeTransfer(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
remote.writeSession(t, "a.jsonl",
|
|
time.Date(2026, 7, 8, 10, 0, 0, 0, time.UTC), "session a")
|
|
_, hs := newMirrorSync(t, remote, t.TempDir())
|
|
var progress []syncpkg.Progress
|
|
hs.Progress = func(p syncpkg.Progress) {
|
|
progress = append(progress, p)
|
|
}
|
|
|
|
prepared, err := hs.Prepare(context.Background())
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { require.NoError(t, prepared.Close()) })
|
|
|
|
assert.Equal(t, []string{
|
|
"Resolving agent directories on devbox",
|
|
"Fetching session manifest from devbox",
|
|
"Compared session manifest from devbox: 1 total, 1 changed, 0 deleted",
|
|
"Downloading session archive from devbox",
|
|
"Extracting session archive from devbox",
|
|
}, uniqueProgressDetails(progress))
|
|
}
|
|
|
|
func TestHTTPSyncLegacyCompressedTransferPreservesWireProgress(t *testing.T) {
|
|
archive := buildHTTPTestTar(t, map[string]string{
|
|
"home/user/.claude/projects/test/session.jsonl": "legacy compressed",
|
|
})
|
|
var compressed bytes.Buffer
|
|
gz := gzip.NewWriter(&compressed)
|
|
_, err := gz.Write(archive)
|
|
require.NoError(t, err)
|
|
require.NoError(t, gz.Close())
|
|
|
|
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
assert.Equal(t, "gzip", r.Header.Get("Accept-Encoding"))
|
|
w.Header().Set("Content-Encoding", "gzip")
|
|
w.Header().Set("Content-Length", strconv.Itoa(compressed.Len()))
|
|
_, _ = w.Write(compressed.Bytes())
|
|
}))
|
|
t.Cleanup(ts.Close)
|
|
|
|
var progress []syncpkg.Progress
|
|
hs := HTTPSync{
|
|
Host: "devbox",
|
|
URL: ts.URL,
|
|
Progress: func(p syncpkg.Progress) {
|
|
progress = append(progress, p)
|
|
},
|
|
}
|
|
root, err := hs.downloadAndExtract(
|
|
context.Background(), ts.Client(), TargetSet{},
|
|
)
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { require.NoError(t, os.RemoveAll(root)) })
|
|
body, err := os.ReadFile(filepath.Join(
|
|
root, "home/user/.claude/projects/test/session.jsonl",
|
|
))
|
|
require.NoError(t, err)
|
|
assert.Equal(t, "legacy compressed", string(body))
|
|
assert.Equal(t, []string{
|
|
"Downloading session archive from devbox",
|
|
"Extracting session archive from devbox",
|
|
}, uniqueProgressDetails(progress))
|
|
assert.Equal(t, int64(compressed.Len()), maxBytesDone(progress))
|
|
assert.Equal(t, int64(compressed.Len()), maxBytesTotal(progress))
|
|
}
|
|
|
|
func TestHTTPSyncRemovesSpooledArchive(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
archive []byte
|
|
contentLength int
|
|
status int
|
|
wantErr bool
|
|
}{
|
|
{
|
|
name: "after successful extraction",
|
|
archive: buildHTTPTestTar(t, map[string]string{
|
|
"home/user/.claude/projects/test/session.jsonl": "complete",
|
|
}),
|
|
},
|
|
{
|
|
name: "after extraction failure",
|
|
archive: tarWithoutEndMarker(t,
|
|
"home/user/.claude/projects/test/session.jsonl", "truncated"),
|
|
wantErr: true,
|
|
},
|
|
{
|
|
name: "after response body read failure",
|
|
archive: buildHTTPTestTar(t, map[string]string{
|
|
"home/user/.claude/projects/test/session.jsonl": "incomplete response",
|
|
}),
|
|
contentLength: 1 << 20,
|
|
wantErr: true,
|
|
},
|
|
{
|
|
name: "after non-success response body read failure",
|
|
archive: []byte("upstream failure"),
|
|
contentLength: 1 << 20,
|
|
status: http.StatusBadGateway,
|
|
wantErr: true,
|
|
},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
|
if tt.contentLength > 0 {
|
|
w.Header().Set("Content-Length", strconv.Itoa(tt.contentLength))
|
|
}
|
|
if tt.status > 0 {
|
|
w.WriteHeader(tt.status)
|
|
}
|
|
_, _ = w.Write(tt.archive)
|
|
}))
|
|
t.Cleanup(ts.Close)
|
|
parent := filepath.Join(t.TempDir(), "remote-mirrors")
|
|
mirrorRoot := filepath.Join(parent, "devbox")
|
|
|
|
err := (HTTPSync{Host: "devbox", URL: ts.URL}).downloadIntoMirror(
|
|
context.Background(), ts.Client(), TargetSet{}, []string{"session.jsonl"},
|
|
false, mirrorRoot,
|
|
)
|
|
if tt.wantErr {
|
|
require.Error(t, err)
|
|
} else {
|
|
require.NoError(t, err)
|
|
}
|
|
if tt.status > 0 {
|
|
var statusErr *StatusError
|
|
require.ErrorAs(t, err, &statusErr)
|
|
assert.Equal(t, tt.status, statusErr.Code)
|
|
}
|
|
entries, readErr := os.ReadDir(parent)
|
|
if errors.Is(readErr, os.ErrNotExist) {
|
|
return
|
|
}
|
|
require.NoError(t, readErr)
|
|
for _, entry := range entries {
|
|
assert.NotContains(t, entry.Name(), "agentsview-http-archive-",
|
|
"owned spool artifacts must be removed")
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestHTTPSyncMirrorRetainsPostExtractionSpoolCleanup(t *testing.T) {
|
|
archive := buildHTTPTestTar(t, map[string]string{
|
|
"home/user/session.jsonl": "complete",
|
|
})
|
|
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
|
_, _ = w.Write(archive)
|
|
}))
|
|
t.Cleanup(ts.Close)
|
|
removeErr := errors.New("spool removal failed")
|
|
removeCalls := 0
|
|
hs := HTTPSync{Host: "devbox", URL: ts.URL, removeArchiveSpool: func(path string) error {
|
|
removeCalls++
|
|
if removeCalls == 1 {
|
|
return removeErr
|
|
}
|
|
return os.RemoveAll(path)
|
|
}}
|
|
mirrorRoot := filepath.Join(t.TempDir(), "mirrors", "devbox")
|
|
|
|
err := hs.downloadIntoMirror(
|
|
context.Background(), ts.Client(), TargetSet{}, []string{"session.jsonl"},
|
|
false, mirrorRoot,
|
|
)
|
|
|
|
require.ErrorIs(t, err, removeErr)
|
|
var owner *downloadedArchiveCleanupError
|
|
require.ErrorAs(t, err, &owner,
|
|
"post-extraction cleanup must remain retryable")
|
|
assert.FileExists(t, filepath.Join(mirrorRoot, "home/user/session.jsonl"))
|
|
require.NoError(t, owner.RetryCleanup())
|
|
assert.Equal(t, 2, removeCalls)
|
|
}
|
|
|
|
func TestHTTPSyncLegacyRetainsCleanupWithoutLeakingExtractedRoot(t *testing.T) {
|
|
archive := buildHTTPTestTar(t, map[string]string{
|
|
"home/user/session.jsonl": "complete",
|
|
})
|
|
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
|
_, _ = w.Write(archive)
|
|
}))
|
|
t.Cleanup(ts.Close)
|
|
tempParent := t.TempDir()
|
|
setPortableTempDir(t, tempParent)
|
|
removeErr := errors.New("spool removal failed")
|
|
removeCalls := 0
|
|
hs := HTTPSync{Host: "devbox", URL: ts.URL, removeArchiveSpool: func(path string) error {
|
|
removeCalls++
|
|
if removeCalls == 1 {
|
|
return removeErr
|
|
}
|
|
return os.RemoveAll(path)
|
|
}}
|
|
|
|
root, err := hs.downloadAndExtract(
|
|
context.Background(), ts.Client(), TargetSet{},
|
|
)
|
|
|
|
assert.Empty(t, root, "failed cleanup must not transfer root ownership")
|
|
require.ErrorIs(t, err, removeErr)
|
|
var owner *downloadedArchiveCleanupError
|
|
require.ErrorAs(t, err, &owner,
|
|
"post-extraction cleanup must remain retryable")
|
|
require.NoError(t, owner.RetryCleanup())
|
|
entries, readErr := os.ReadDir(tempParent)
|
|
require.NoError(t, readErr)
|
|
assert.Empty(t, entries,
|
|
"retry must release both the spool and extracted legacy root")
|
|
}
|
|
|
|
func TestHTTPSyncLegacyRetainsSpoolWhenExtractionRootCreationFails(t *testing.T) {
|
|
archive := buildHTTPTestTar(t, map[string]string{
|
|
"home/user/session.jsonl": "complete",
|
|
})
|
|
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
|
w.Header().Set("Content-Length", strconv.Itoa(len(archive)))
|
|
_, _ = w.Write(archive)
|
|
}))
|
|
t.Cleanup(ts.Close)
|
|
validTemp := t.TempDir()
|
|
setPortableTempDir(t, validTemp)
|
|
invalidTemp := filepath.Join(t.TempDir(), "not-a-directory")
|
|
require.NoError(t, os.WriteFile(invalidTemp, []byte("block"), 0o600))
|
|
removeErr := errors.New("spool cleanup failed")
|
|
removeCalls := 0
|
|
hs := HTTPSync{
|
|
Host: "devbox",
|
|
URL: ts.URL,
|
|
Progress: func(p syncpkg.Progress) {
|
|
if p.BytesDone == int64(len(archive)) {
|
|
for _, name := range []string{"TMPDIR", "TMP", "TEMP"} {
|
|
require.NoError(t, os.Setenv(name, invalidTemp))
|
|
}
|
|
}
|
|
},
|
|
removeArchiveSpool: func(string) error {
|
|
removeCalls++
|
|
if removeCalls == 1 {
|
|
return removeErr
|
|
}
|
|
return nil
|
|
},
|
|
}
|
|
|
|
root, err := hs.downloadAndExtract(
|
|
context.Background(), ts.Client(), TargetSet{},
|
|
)
|
|
|
|
assert.Empty(t, root)
|
|
require.ErrorContains(t, err, "create temp dir")
|
|
require.ErrorIs(t, err, removeErr)
|
|
var owner *downloadedArchiveCleanupError
|
|
require.ErrorAs(t, err, &owner,
|
|
"failed cleanup must retain the downloaded spool")
|
|
require.NoError(t, owner.RetryCleanup())
|
|
assert.Equal(t, 2, removeCalls)
|
|
}
|
|
|
|
func TestHTTPSyncRemovesSpooledArchiveOnCancellation(t *testing.T) {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
parent := filepath.Join(t.TempDir(), "remote-mirrors")
|
|
resp := &http.Response{
|
|
StatusCode: http.StatusOK,
|
|
Body: &cancelEOFBody{ctx: ctx},
|
|
Header: make(http.Header),
|
|
}
|
|
hs := HTTPSync{Progress: func(p syncpkg.Progress) {
|
|
if p.BytesDone > 0 {
|
|
cancel()
|
|
}
|
|
}}
|
|
|
|
archive, err := hs.downloadArchive(ctx, resp, "Downloading", parent)
|
|
if archive != nil {
|
|
t.Cleanup(func() { require.NoError(t, archive.Close()) })
|
|
}
|
|
require.ErrorIs(t, err, context.Canceled)
|
|
entries, readErr := os.ReadDir(parent)
|
|
require.NoError(t, readErr)
|
|
assert.Empty(t, entries)
|
|
}
|
|
|
|
func TestDownloadArchiveCloseFailureRemovesIsolatedSpool(t *testing.T) {
|
|
closeErr := errors.New("close response body")
|
|
parent := t.TempDir()
|
|
require.NoError(t, os.WriteFile(
|
|
filepath.Join(parent, "existing"), []byte("keep"), 0o644,
|
|
))
|
|
resp := &http.Response{
|
|
StatusCode: http.StatusOK,
|
|
Body: &closeErrorBody{
|
|
Reader: bytes.NewReader([]byte("wire archive")),
|
|
err: closeErr,
|
|
},
|
|
Header: make(http.Header),
|
|
}
|
|
|
|
archive, err := (HTTPSync{}).downloadArchive(
|
|
context.Background(), resp, "Downloading", parent,
|
|
)
|
|
|
|
assert.Nil(t, archive)
|
|
require.ErrorIs(t, err, closeErr)
|
|
entries, readErr := os.ReadDir(parent)
|
|
require.NoError(t, readErr)
|
|
require.Len(t, entries, 1)
|
|
assert.Equal(t, "existing", entries[0].Name())
|
|
}
|
|
|
|
func TestDownloadedArchiveUsesIsolatedSpoolOutsideMirror(t *testing.T) {
|
|
parent := t.TempDir()
|
|
mirrorRoot := filepath.Join(parent, "devbox")
|
|
require.NoError(t, os.Mkdir(mirrorRoot, 0o755))
|
|
require.NoError(t, os.WriteFile(
|
|
filepath.Join(parent, "existing"), []byte("keep"), 0o644,
|
|
))
|
|
resp := &http.Response{
|
|
StatusCode: http.StatusOK,
|
|
Body: io.NopCloser(bytes.NewReader([]byte("wire archive"))),
|
|
Header: make(http.Header),
|
|
}
|
|
|
|
archive, err := (HTTPSync{}).downloadArchive(
|
|
context.Background(), resp, "Downloading", parent,
|
|
)
|
|
require.NoError(t, err)
|
|
require.NotNil(t, archive)
|
|
t.Cleanup(func() { require.NoError(t, archive.Close()) })
|
|
|
|
tempDir := filepath.Dir(archive.path)
|
|
assert.DirExists(t, tempDir)
|
|
assert.FileExists(t, archive.path)
|
|
assert.Equal(t, parent, filepath.Dir(tempDir))
|
|
assert.NotEqual(t, parent, tempDir)
|
|
assert.False(t, within(mirrorRoot, archive.path),
|
|
"spool must not be created inside the manifest mirror")
|
|
require.NoError(t, archive.Close())
|
|
entries, readErr := os.ReadDir(parent)
|
|
require.NoError(t, readErr)
|
|
require.Len(t, entries, 2)
|
|
assert.Equal(t, []string{"devbox", "existing"},
|
|
[]string{entries[0].Name(), entries[1].Name()})
|
|
}
|
|
|
|
func TestDownloadedArchiveExtractionCancellationMidEntry(t *testing.T) {
|
|
body := make([]byte, 32<<10)
|
|
state := uint32(1)
|
|
for i := range body {
|
|
state ^= state << 13
|
|
state ^= state >> 17
|
|
state ^= state << 5
|
|
body[i] = byte(state)
|
|
}
|
|
archiveBytes := buildHTTPTestTar(t, map[string]string{
|
|
"home/user/large-session.jsonl": string(body),
|
|
})
|
|
var compressed bytes.Buffer
|
|
gz := gzip.NewWriter(&compressed)
|
|
_, err := gz.Write(archiveBytes)
|
|
require.NoError(t, err)
|
|
require.NoError(t, gz.Close())
|
|
require.Greater(t, compressed.Len(), 1024)
|
|
|
|
tests := []struct {
|
|
name string
|
|
wire []byte
|
|
compressed bool
|
|
}{
|
|
{name: "plain", wire: archiveBytes},
|
|
{name: "gzip", wire: compressed.Bytes(), compressed: true},
|
|
}
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
spoolParent := t.TempDir()
|
|
spoolDir := filepath.Join(spoolParent, "agentsview-http-archive-test")
|
|
require.NoError(t, os.Mkdir(spoolDir, 0o700))
|
|
require.NoError(t, os.WriteFile(
|
|
filepath.Join(spoolDir, "owned"), []byte("spool"), 0o600,
|
|
))
|
|
reached := make(chan struct{})
|
|
release := make(chan struct{})
|
|
reader := &gatedArchiveReader{
|
|
r: bytes.NewReader(tt.wire),
|
|
gateAfter: 768,
|
|
reached: reached,
|
|
release: release,
|
|
}
|
|
archive := &downloadedArchive{
|
|
dir: spoolDir,
|
|
path: filepath.Join(spoolDir, "archive"),
|
|
compressed: tt.compressed,
|
|
open: func() (io.ReadCloser, error) { return reader, nil },
|
|
}
|
|
t.Cleanup(func() { require.NoError(t, archive.Close()) })
|
|
dst := filepath.Join(t.TempDir(), "extracted")
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
t.Cleanup(cancel)
|
|
released := false
|
|
t.Cleanup(func() {
|
|
if !released {
|
|
close(release)
|
|
}
|
|
})
|
|
errCh := make(chan error, 1)
|
|
go func() {
|
|
errCh <- archive.extract(ctx, dst, nil, "Extracting")
|
|
}()
|
|
|
|
select {
|
|
case <-reached:
|
|
case <-time.After(time.Second):
|
|
require.FailNow(t, "timed out waiting for extraction to enter file body")
|
|
}
|
|
cancel()
|
|
close(release)
|
|
released = true
|
|
select {
|
|
case err := <-errCh:
|
|
require.ErrorIs(t, err, context.Canceled)
|
|
case <-time.After(time.Second):
|
|
require.FailNow(t, "timed out waiting for canceled extraction")
|
|
}
|
|
assert.NoFileExists(t,
|
|
filepath.Join(dst, "home/user/large-session.jsonl"))
|
|
require.NoError(t, archive.Close())
|
|
assert.NoDirExists(t, spoolDir)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestPrepareHTTPSyncCancellationReleasesMirrorOwnership(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
remote.writeSession(t, "large.jsonl",
|
|
time.Date(2026, 7, 11, 12, 0, 0, 0, time.UTC),
|
|
strings.Repeat("entry\n", 4096))
|
|
dataDir := t.TempDir()
|
|
_, hs := newMirrorSync(t, remote, dataDir)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
extractionStarted := false
|
|
hs.Progress = func(p syncpkg.Progress) {
|
|
if strings.HasPrefix(p.Detail, "Extracting ") {
|
|
extractionStarted = true
|
|
cancel()
|
|
}
|
|
}
|
|
|
|
prepared, err := hs.Prepare(ctx)
|
|
|
|
assert.Nil(t, prepared)
|
|
require.ErrorIs(t, err, context.Canceled)
|
|
assert.True(t, extractionStarted)
|
|
mirrorRoot := MirrorDir(dataDir, "devbox")
|
|
assertMirrorUnlocked(t, mirrorRoot)
|
|
entries, readErr := os.ReadDir(filepath.Dir(mirrorRoot))
|
|
require.NoError(t, readErr)
|
|
for _, entry := range entries {
|
|
assert.NotContains(t, entry.Name(), "agentsview-http-archive-",
|
|
"canceled preparation must release its spool ownership")
|
|
}
|
|
}
|
|
|
|
func TestHTTPSyncLegacyCancellationCleansOwnedRoots(t *testing.T) {
|
|
archive := buildHTTPTestTar(t, map[string]string{
|
|
"home/user/large-session.jsonl": strings.Repeat("entry\n", 4096),
|
|
})
|
|
var compressed bytes.Buffer
|
|
gz := gzip.NewWriter(&compressed)
|
|
_, err := gz.Write(archive)
|
|
require.NoError(t, err)
|
|
require.NoError(t, gz.Close())
|
|
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
|
w.Header().Set("Content-Encoding", "gzip")
|
|
w.Header().Set("Content-Length", strconv.Itoa(compressed.Len()))
|
|
_, _ = w.Write(compressed.Bytes())
|
|
}))
|
|
t.Cleanup(ts.Close)
|
|
tempParent := t.TempDir()
|
|
setPortableTempDir(t, tempParent)
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
hs := HTTPSync{
|
|
Host: "devbox",
|
|
URL: ts.URL,
|
|
Progress: func(p syncpkg.Progress) {
|
|
if strings.HasPrefix(p.Detail, "Extracting ") {
|
|
cancel()
|
|
}
|
|
},
|
|
}
|
|
|
|
root, err := hs.downloadAndExtract(ctx, ts.Client(), TargetSet{})
|
|
|
|
assert.Empty(t, root)
|
|
require.ErrorIs(t, err, context.Canceled)
|
|
entries, readErr := os.ReadDir(tempParent)
|
|
require.NoError(t, readErr)
|
|
assert.Empty(t, entries,
|
|
"canceled legacy extraction must remove its spool and extraction root")
|
|
}
|
|
|
|
type cancelEOFBody struct {
|
|
ctx context.Context
|
|
sent bool
|
|
}
|
|
|
|
func (b *cancelEOFBody) Read(p []byte) (int, error) {
|
|
if !b.sent {
|
|
b.sent = true
|
|
return copy(p, "partial response"), nil
|
|
}
|
|
<-b.ctx.Done()
|
|
return 0, io.EOF
|
|
}
|
|
|
|
func (*cancelEOFBody) Close() error { return nil }
|
|
|
|
type closeErrorBody struct {
|
|
io.Reader
|
|
err error
|
|
}
|
|
|
|
func (b *closeErrorBody) Close() error { return b.err }
|
|
|
|
type gatedArchiveReader struct {
|
|
r *bytes.Reader
|
|
gateAfter int
|
|
read int
|
|
reached chan struct{}
|
|
release chan struct{}
|
|
gated bool
|
|
}
|
|
|
|
func (r *gatedArchiveReader) Read(p []byte) (int, error) {
|
|
if r.read >= r.gateAfter && !r.gated {
|
|
r.gated = true
|
|
close(r.reached)
|
|
<-r.release
|
|
}
|
|
if len(p) > 256 {
|
|
p = p[:256]
|
|
}
|
|
n, err := r.r.Read(p)
|
|
r.read += n
|
|
return n, err
|
|
}
|
|
|
|
func (*gatedArchiveReader) Close() error { return nil }
|
|
|
|
func progressDetails(progress []syncpkg.Progress) []string {
|
|
out := make([]string, 0, len(progress))
|
|
for _, p := range progress {
|
|
if p.Detail != "" {
|
|
out = append(out, p.Detail)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
func uniqueProgressDetails(progress []syncpkg.Progress) []string {
|
|
var out []string
|
|
for _, detail := range progressDetails(progress) {
|
|
if len(out) == 0 || out[len(out)-1] != detail {
|
|
out = append(out, detail)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
func maxBytesDone(progress []syncpkg.Progress) int64 {
|
|
var max int64
|
|
for _, p := range progress {
|
|
if p.BytesDone > max {
|
|
max = p.BytesDone
|
|
}
|
|
}
|
|
return max
|
|
}
|
|
|
|
func maxBytesTotal(progress []syncpkg.Progress) int64 {
|
|
var max int64
|
|
for _, p := range progress {
|
|
if p.BytesTotal > max {
|
|
max = p.BytesTotal
|
|
}
|
|
}
|
|
return max
|
|
}
|
|
|
|
func buildHTTPTestTar(t *testing.T, files map[string]string) []byte {
|
|
t.Helper()
|
|
var buf bytes.Buffer
|
|
tw := tar.NewWriter(&buf)
|
|
mtime := time.Date(2026, 6, 28, 12, 0, 0, 0, time.UTC)
|
|
for name, body := range files {
|
|
hdr := &tar.Header{
|
|
Name: name,
|
|
Mode: 0o644,
|
|
Size: int64(len(body)),
|
|
ModTime: mtime,
|
|
}
|
|
require.NoError(t, tw.WriteHeader(hdr))
|
|
_, err := tw.Write([]byte(body))
|
|
require.NoError(t, err)
|
|
}
|
|
require.NoError(t, tw.Close())
|
|
return buf.Bytes()
|
|
}
|
|
|
|
// mirrorTestRemote is a fake remote daemon backed by a real directory
|
|
// tree, serving targets/manifest/archive with the same package
|
|
// functions the real server uses.
|
|
type mirrorTestRemote struct {
|
|
dir string // remote-side agent dir (absolute)
|
|
targets TargetSet
|
|
archiveRequests []ArchiveRequest
|
|
archiveBody []byte
|
|
onArchive func(ArchiveRequest)
|
|
manifestStatus int // 0 = serve manifest; else respond with this status
|
|
manifestHTML bool // true = mimic an old daemon's SPA catch-all
|
|
onManifest func() // called before serving a manifest response
|
|
onManifestRequest func(*http.Request)
|
|
rejectDelta bool
|
|
ts *httptest.Server
|
|
}
|
|
|
|
func newMirrorTestRemote(t *testing.T) *mirrorTestRemote {
|
|
t.Helper()
|
|
remote := &mirrorTestRemote{
|
|
dir: filepath.Join(t.TempDir(), "claude-projects"),
|
|
}
|
|
require.NoError(t, os.MkdirAll(remote.dir, 0o755))
|
|
remote.targets = TargetSet{
|
|
Dirs: map[parser.AgentType][]string{parser.AgentClaude: {remote.dir}},
|
|
}
|
|
remote.ts = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
switch r.URL.Path {
|
|
case "/api/v1/remote-sync/targets":
|
|
w.Header().Set("Content-Type", "application/json")
|
|
require.NoError(t, json.NewEncoder(w).Encode(remote.targets))
|
|
case "/api/v1/remote-sync/manifest":
|
|
if remote.onManifest != nil {
|
|
remote.onManifest()
|
|
}
|
|
if remote.onManifestRequest != nil {
|
|
remote.onManifestRequest(r)
|
|
}
|
|
if remote.manifestStatus != 0 {
|
|
http.Error(w, "no manifest here", remote.manifestStatus)
|
|
return
|
|
}
|
|
if remote.manifestHTML {
|
|
w.Header().Set("Content-Type", "text/html; charset=utf-8")
|
|
_, _ = w.Write([]byte(
|
|
"<!doctype html><html><body>spa</body></html>"))
|
|
return
|
|
}
|
|
// Mirror the real handler: build the manifest from the
|
|
// requested targets and refuse file-scoped agents.
|
|
var req TargetSet
|
|
require.NoError(t, json.NewDecoder(r.Body).Decode(&req))
|
|
if req.HasFileScopedAgents() {
|
|
http.Error(w, "manifest unavailable for file-scoped agents",
|
|
http.StatusNotImplemented)
|
|
return
|
|
}
|
|
manifest, err := BuildManifest(req)
|
|
require.NoError(t, err)
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.Header().Set("Content-Encoding", "gzip")
|
|
gz := gzip.NewWriter(w)
|
|
require.NoError(t, json.NewEncoder(gz).Encode(manifest))
|
|
require.NoError(t, gz.Close())
|
|
case "/api/v1/remote-sync/archive":
|
|
var req ArchiveRequest
|
|
require.NoError(t, json.NewDecoder(r.Body).Decode(&req))
|
|
remote.archiveRequests = append(remote.archiveRequests, req)
|
|
if remote.onArchive != nil {
|
|
remote.onArchive(req)
|
|
}
|
|
if remote.rejectDelta && req.DeltaFiles != nil {
|
|
http.Error(w, "delta not allowed", http.StatusForbidden)
|
|
return
|
|
}
|
|
// Serve the requested target subset, as the real handler
|
|
// does after SelectAllowedTargets.
|
|
w.Header().Set("Content-Type", "application/x-tar")
|
|
if remote.archiveBody != nil {
|
|
_, _ = w.Write(remote.archiveBody)
|
|
return
|
|
}
|
|
if req.DeltaFiles != nil {
|
|
require.NoError(t, WriteArchiveFiles(
|
|
w, remote.targets.DeltaAllowedRoots(), req.DeltaFiles))
|
|
} else {
|
|
require.NoError(t, WriteArchive(w, req.TargetSet))
|
|
}
|
|
default:
|
|
http.NotFound(w, r)
|
|
}
|
|
}))
|
|
t.Cleanup(remote.ts.Close)
|
|
return remote
|
|
}
|
|
|
|
// addFileScopedAgent registers a second agent whose targets are file
|
|
// scoped, mimicking Windsurf's curated export, and returns the remote
|
|
// file path. The file's content never parses as a session; partition
|
|
// tests assert transfer behavior, and Windsurf import correctness is
|
|
// covered by the server and sync tests.
|
|
func (r *mirrorTestRemote) addFileScopedAgent(t *testing.T) string {
|
|
t.Helper()
|
|
dir := filepath.Join(filepath.Dir(r.dir), "scoped-agent")
|
|
require.NoError(t, os.MkdirAll(dir, 0o755))
|
|
path := filepath.Join(dir, "export.txt")
|
|
require.NoError(t, os.WriteFile(path, []byte("file-scoped export\n"), 0o644))
|
|
r.targets.Dirs[parser.AgentGemini] = []string{dir}
|
|
r.targets.Files = map[parser.AgentType][]string{
|
|
parser.AgentGemini: {path},
|
|
}
|
|
return path
|
|
}
|
|
|
|
// writeSession writes a session file with one user message per text.
|
|
// Message timestamps are deterministic, so writing the same file again
|
|
// with the previous texts plus new ones is a byte-identical prefix
|
|
// append — the realistic mutation for JSONL session files, and one the
|
|
// engine's incremental parse handles. (In-place rewrites that grow are
|
|
// misread as appends for remote paths — pre-existing engine gap, out
|
|
// of scope here.)
|
|
func (r *mirrorTestRemote) writeSession(
|
|
t *testing.T, name string, mtime time.Time, userTexts ...string,
|
|
) string {
|
|
t.Helper()
|
|
dir := filepath.Join(r.dir, "test-project")
|
|
require.NoError(t, os.MkdirAll(dir, 0o755))
|
|
builder := testjsonl.NewSessionBuilder()
|
|
for i, text := range userTexts {
|
|
ts := time.Date(2024, 1, 1, 0, i, 0, 0, time.UTC).Format(time.RFC3339)
|
|
builder = builder.AddClaudeUser(ts, text)
|
|
}
|
|
path := filepath.Join(dir, name)
|
|
require.NoError(t, os.WriteFile(path, []byte(builder.String()), 0o644))
|
|
require.NoError(t, os.Chtimes(path, mtime, mtime))
|
|
return path
|
|
}
|
|
|
|
func newMirrorSync(t *testing.T, remote *mirrorTestRemote, dataDir string) (*db.DB, HTTPSync) {
|
|
t.Helper()
|
|
database, err := db.Open(filepath.Join(t.TempDir(), "test.db"))
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { require.NoError(t, database.Close()) })
|
|
return database, HTTPSync{
|
|
Host: "devbox",
|
|
URL: remote.ts.URL,
|
|
Token: "remote-token",
|
|
DataDir: dataDir,
|
|
DB: database,
|
|
}
|
|
}
|
|
|
|
func TestHTTPSyncMirrorSecondSyncTransfersOnlyDelta(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
base := time.Date(2026, 7, 8, 10, 0, 0, 123456789, time.UTC)
|
|
remote.writeSession(t, "a.jsonl", base, "session a")
|
|
staleRemote := remote.writeSession(t, "b.jsonl", base, "session b")
|
|
remote.writeSession(t, "c.jsonl", base, "session c")
|
|
remote.writeSession(t, "d.jsonl", base, "session d")
|
|
remote.writeSession(t, "e.jsonl", base, "session e")
|
|
dataDir := t.TempDir()
|
|
database, hs := newMirrorSync(t, remote, dataDir)
|
|
|
|
stats, err := hs.Run(context.Background())
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 5, stats.SessionsSynced)
|
|
require.Len(t, remote.archiveRequests, 1)
|
|
assert.Empty(t, remote.archiveRequests[0].DeltaFiles, "bootstrap uses the full archive")
|
|
|
|
// Append to one, add one, delete one on the remote. The fetch set
|
|
// (2 of the 5 files now in the manifest) stays under the bootstrap
|
|
// heuristic's half-corpus threshold, so this sync must go delta.
|
|
changed := remote.writeSession(t, "a.jsonl", base.Add(5*time.Second),
|
|
"session a", "session a continued")
|
|
added := remote.writeSession(t, "f.jsonl", base.Add(6*time.Second), "session f")
|
|
require.NoError(t, os.Remove(staleRemote))
|
|
|
|
stats, err = hs.Run(context.Background())
|
|
require.NoError(t, err)
|
|
require.Len(t, remote.archiveRequests, 2)
|
|
assert.ElementsMatch(t, []string{changed, added}, remote.archiveRequests[1].DeltaFiles)
|
|
assert.Equal(t, 2, stats.SessionsSynced)
|
|
|
|
// The deleted remote file is gone from the mirror, but its
|
|
// session survives in the DB (archive semantics).
|
|
mirrorRoot := MirrorDir(dataDir, "devbox")
|
|
staleLocal, err := safeRemappedRemotePath(mirrorRoot, staleRemote)
|
|
require.NoError(t, err)
|
|
assert.NoFileExists(t, staleLocal)
|
|
page, err := database.ListSessions(context.Background(), db.SessionFilter{Limit: 10})
|
|
require.NoError(t, err)
|
|
assert.Len(t, page.Sessions, 6)
|
|
|
|
// Third sync with no remote changes: no archive request at all.
|
|
stats, err = hs.Run(context.Background())
|
|
require.NoError(t, err)
|
|
assert.Len(t, remote.archiveRequests, 2)
|
|
assert.Equal(t, 0, stats.SessionsSynced)
|
|
}
|
|
|
|
func TestHTTPSyncFallsBackToLegacyWhenManifestMissing(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
remote.manifestStatus = http.StatusNotFound
|
|
remote.writeSession(t, "a.jsonl",
|
|
time.Date(2026, 7, 8, 10, 0, 0, 0, time.UTC), "legacy fallback")
|
|
dataDir := t.TempDir()
|
|
_, hs := newMirrorSync(t, remote, dataDir)
|
|
|
|
stats, err := hs.Run(context.Background())
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 1, stats.SessionsSynced)
|
|
require.Len(t, remote.archiveRequests, 1)
|
|
assert.Empty(t, remote.archiveRequests[0].DeltaFiles)
|
|
assert.NoDirExists(t, MirrorDir(dataDir, "devbox"))
|
|
}
|
|
|
|
func TestHTTPSyncFallsBackToLegacyWhenManifestServesSPAHTML(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
remote.writeSession(t, "a.jsonl",
|
|
time.Date(2026, 7, 8, 10, 0, 0, 0, time.UTC), "session a")
|
|
remote.manifestHTML = true
|
|
_, hs := newMirrorSync(t, remote, t.TempDir())
|
|
|
|
stats, err := hs.Run(context.Background())
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 1, stats.SessionsSynced)
|
|
require.Len(t, remote.archiveRequests, 1)
|
|
assert.Empty(t, remote.archiveRequests[0].DeltaFiles)
|
|
}
|
|
|
|
func TestHTTPSyncFallsBackToFullWhenDeltaRejected(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
base := time.Date(2026, 7, 8, 10, 0, 0, 0, time.UTC)
|
|
remote.writeSession(t, "a.jsonl", base, "session a")
|
|
remote.writeSession(t, "b.jsonl", base, "session b")
|
|
remote.writeSession(t, "c.jsonl", base, "session c")
|
|
dataDir := t.TempDir()
|
|
_, hs := newMirrorSync(t, remote, dataDir)
|
|
|
|
_, err := hs.Run(context.Background())
|
|
require.NoError(t, err)
|
|
|
|
remote.rejectDelta = true
|
|
remote.writeSession(t, "a.jsonl", base.Add(5*time.Second),
|
|
"session a", "session a continued")
|
|
|
|
stats, err := hs.Run(context.Background())
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 1, stats.SessionsSynced)
|
|
// Requests: bootstrap full, rejected delta, retried full.
|
|
require.Len(t, remote.archiveRequests, 3)
|
|
assert.NotEmpty(t, remote.archiveRequests[1].DeltaFiles)
|
|
assert.Empty(t, remote.archiveRequests[2].DeltaFiles)
|
|
}
|
|
|
|
func TestHTTPSyncIncrementalMatchesFreshFullSync(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
base := time.Date(2026, 7, 8, 10, 0, 0, 555666777, time.UTC)
|
|
// Enough unchanged files that the second sync's two changed files
|
|
// stay under the half-corpus bootstrap heuristic and exercise the
|
|
// delta path rather than a full re-download.
|
|
appended := remote.writeSession(t, "a.jsonl", base, "session a")
|
|
remote.writeSession(t, "b.jsonl", base, "session b")
|
|
remote.writeSession(t, "d.jsonl", base, "session d")
|
|
remote.writeSession(t, "e.jsonl", base, "session e")
|
|
|
|
incDB, incSync := newMirrorSync(t, remote, t.TempDir())
|
|
_, err := incSync.Run(context.Background())
|
|
require.NoError(t, err)
|
|
|
|
remote.writeSession(t, "a.jsonl", base.Add(2*time.Second),
|
|
"session a", "session a continued")
|
|
added := remote.writeSession(t, "c.jsonl", base.Add(3*time.Second), "session c")
|
|
_, err = incSync.Run(context.Background())
|
|
require.NoError(t, err)
|
|
|
|
// Requests: bootstrap full, then a delta for exactly the changes.
|
|
require.Len(t, remote.archiveRequests, 2)
|
|
assert.Empty(t, remote.archiveRequests[0].DeltaFiles)
|
|
assert.ElementsMatch(t, []string{appended, added},
|
|
remote.archiveRequests[1].DeltaFiles)
|
|
|
|
freshDB, freshSync := newMirrorSync(t, remote, t.TempDir())
|
|
_, err = freshSync.Run(context.Background())
|
|
require.NoError(t, err)
|
|
|
|
assert.Equal(t, sessionSummaries(t, freshDB), sessionSummaries(t, incDB))
|
|
}
|
|
|
|
func TestHTTPSyncMirrorRecoversFromDirAtFilePath(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
base := time.Date(2026, 7, 8, 10, 0, 0, 123456789, time.UTC)
|
|
wedged := remote.writeSession(t, "a.jsonl", base, "session a")
|
|
remote.writeSession(t, "b.jsonl", base, "session b")
|
|
remote.writeSession(t, "c.jsonl", base, "session c")
|
|
remote.writeSession(t, "d.jsonl", base, "session d")
|
|
remote.writeSession(t, "e.jsonl", base, "session e")
|
|
dataDir := t.TempDir()
|
|
_, hs := newMirrorSync(t, remote, dataDir)
|
|
|
|
_, err := hs.Run(context.Background())
|
|
require.NoError(t, err)
|
|
|
|
// Simulate a crashed extraction: a directory occupies a.jsonl's
|
|
// mirror path (MkdirAll ran, the file write never happened).
|
|
local, err := safeRemappedRemotePath(MirrorDir(dataDir, "devbox"), wedged)
|
|
require.NoError(t, err)
|
|
require.NoError(t, os.Remove(local))
|
|
require.NoError(t, os.Mkdir(local, 0o755))
|
|
|
|
_, err = hs.Run(context.Background())
|
|
require.NoError(t, err)
|
|
info, err := os.Stat(local)
|
|
require.NoError(t, err)
|
|
assert.True(t, info.Mode().IsRegular())
|
|
// Recovery re-fetched only the wedged file, via the delta path.
|
|
require.Len(t, remote.archiveRequests, 2)
|
|
assert.Equal(t, []string{wedged}, remote.archiveRequests[1].DeltaFiles)
|
|
}
|
|
|
|
// sessionSummaries returns a sorted, comparable projection of every
|
|
// session: identity plus message count, which changes when a session
|
|
// file's new content is parsed.
|
|
func sessionSummaries(t *testing.T, database *db.DB) []string {
|
|
t.Helper()
|
|
page, err := database.ListSessions(context.Background(), db.SessionFilter{Limit: 100})
|
|
require.NoError(t, err)
|
|
out := make([]string, 0, len(page.Sessions))
|
|
for _, s := range page.Sessions {
|
|
count, ok := database.GetSessionMessageCount(s.ID)
|
|
require.True(t, ok, "message count for %s", s.ID)
|
|
hash, ok := database.GetSessionFileHash(s.ID)
|
|
require.True(t, ok, "file hash for %s", s.ID)
|
|
out = append(out, fmt.Sprintf("%s|%s|%d|%s", s.ID, s.Machine, count, hash))
|
|
}
|
|
sort.Strings(out)
|
|
return out
|
|
}
|
|
|
|
// A host with a file-scoped agent (Windsurf) must not lose incremental
|
|
// sync for its dir-scoped corpus: the manifest request carries only the
|
|
// dir-scoped targets, and the file-scoped exports arrive as a separate
|
|
// small full archive on every sync.
|
|
func TestHTTPSyncMirrorPartitionsFileScopedAgents(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
base := time.Date(2026, 7, 8, 10, 0, 0, 123456789, time.UTC)
|
|
remote.writeSession(t, "a.jsonl", base, "session a")
|
|
remote.writeSession(t, "b.jsonl", base, "session b")
|
|
remote.writeSession(t, "c.jsonl", base, "session c")
|
|
remote.writeSession(t, "d.jsonl", base, "session d")
|
|
remote.writeSession(t, "e.jsonl", base, "session e")
|
|
scoped := remote.addFileScopedAgent(t)
|
|
dataDir := t.TempDir()
|
|
_, hs := newMirrorSync(t, remote, dataDir)
|
|
|
|
stats, err := hs.Run(context.Background())
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 5, stats.SessionsSynced)
|
|
// Bootstrap: one full archive for the dir-scoped corpus, one for
|
|
// the file-scoped agent — never a combined whole-host archive.
|
|
require.Len(t, remote.archiveRequests, 2)
|
|
assert.Nil(t, remote.archiveRequests[0].DeltaFiles)
|
|
assert.Contains(t, remote.archiveRequests[0].Dirs, parser.AgentClaude)
|
|
assert.NotContains(t, remote.archiveRequests[0].Dirs, parser.AgentGemini)
|
|
assert.Contains(t, remote.archiveRequests[1].Files, parser.AgentGemini)
|
|
assert.NotContains(t, remote.archiveRequests[1].Dirs, parser.AgentClaude)
|
|
|
|
mirrorRoot := MirrorDir(dataDir, "devbox")
|
|
scopedLocal, err := safeRemappedRemotePath(mirrorRoot, scoped)
|
|
require.NoError(t, err)
|
|
assert.FileExists(t, scopedLocal)
|
|
|
|
// Second sync: the changed session travels as a delta, and the
|
|
// file-scoped export — cleared by the mirror deletion pass because
|
|
// it is never in the manifest — is re-fetched in full and lands
|
|
// back in the mirror.
|
|
changed := remote.writeSession(t, "a.jsonl", base.Add(5*time.Second),
|
|
"session a", "session a continued")
|
|
stats, err = hs.Run(context.Background())
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 1, stats.SessionsSynced)
|
|
require.Len(t, remote.archiveRequests, 4)
|
|
assert.Equal(t, []string{changed}, remote.archiveRequests[2].DeltaFiles)
|
|
assert.Contains(t, remote.archiveRequests[3].Files, parser.AgentGemini)
|
|
assert.FileExists(t, scopedLocal)
|
|
|
|
// The file-scoped export disappears from the remote: the deletion
|
|
// pass clears its mirror copy and nothing re-populates it, matching
|
|
// the legacy path where only the current export was ever extracted.
|
|
require.NoError(t, os.Remove(scoped))
|
|
delete(remote.targets.Dirs, parser.AgentGemini)
|
|
remote.targets.Files = nil
|
|
_, err = hs.Run(context.Background())
|
|
require.NoError(t, err)
|
|
assert.NoFileExists(t, scopedLocal)
|
|
assert.Len(t, remote.archiveRequests, 4,
|
|
"no archive requests when nothing changed and no file-scoped targets remain")
|
|
}
|
|
|
|
// The mirror lock must already be held when the manifest is fetched:
|
|
// otherwise two concurrent syncs can fetch manifests in one order and
|
|
// apply them in another, and the stale manifest's deletion pass
|
|
// removes files the newer sync just mirrored.
|
|
func TestHTTPSyncHoldsMirrorLockDuringManifestFetch(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
remote.writeSession(t, "a.jsonl",
|
|
time.Date(2026, 7, 8, 10, 0, 0, 0, time.UTC), "session a")
|
|
dataDir := t.TempDir()
|
|
_, hs := newMirrorSync(t, remote, dataDir)
|
|
|
|
mirrorRoot := MirrorDir(dataDir, "devbox")
|
|
remote.onManifest = func() {
|
|
// assert, not require: this runs on the server goroutine, where
|
|
// FailNow must not be called.
|
|
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
|
|
defer cancel()
|
|
lock, err := AcquireMirrorLock(ctx, mirrorRoot)
|
|
if lock != nil {
|
|
// On regression the acquire succeeds; release immediately so
|
|
// the sync under test fails on the assertion instead of
|
|
// deadlocking against a leaked lock.
|
|
_ = lock.Close()
|
|
}
|
|
assert.Error(t, err, "mirror lock must be held during the manifest fetch")
|
|
}
|
|
|
|
_, err := hs.Run(context.Background())
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
func TestPrepareHTTPSyncContributorDeltaCardinality(t *testing.T) {
|
|
for _, fileCount := range []int{5, 500} {
|
|
t.Run(strconv.Itoa(fileCount), func(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
base := time.Date(2026, 7, 8, 10, 0, 0, 123456789, time.UTC)
|
|
var changed string
|
|
for i := range fileCount {
|
|
path := remote.writeSession(t, fmt.Sprintf("%03d.jsonl", i),
|
|
base, fmt.Sprintf("session %d", i))
|
|
if i == 0 {
|
|
changed = path
|
|
}
|
|
}
|
|
dataDir := t.TempDir()
|
|
database, hs := newMirrorSync(t, remote, dataDir)
|
|
|
|
prepared, err := hs.Prepare(context.Background())
|
|
require.NoError(t, err)
|
|
require.NoError(t, prepared.Close())
|
|
require.Len(t, remote.archiveRequests, 1)
|
|
assert.Nil(t, remote.archiveRequests[0].DeltaFiles)
|
|
|
|
remote.writeSession(t, "000.jsonl", base.Add(time.Second),
|
|
"session 0", "session 0 continued")
|
|
hs.Full = true
|
|
prepared, err = hs.Prepare(context.Background())
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { require.NoError(t, prepared.Close()) })
|
|
contributor, err := prepared.RebuildContributor()
|
|
require.NoError(t, err)
|
|
engine := syncpkg.NewEngine(database, syncpkg.EngineConfig{})
|
|
stats, err := engine.ResyncAllWithOptions(
|
|
context.Background(), nil,
|
|
syncpkg.RebuildOptions{Contributors: []syncpkg.RebuildContributor{
|
|
contributor,
|
|
}},
|
|
)
|
|
engine.Close()
|
|
require.NoError(t, err)
|
|
assert.False(t, stats.Aborted)
|
|
require.NoError(t, prepared.Close())
|
|
|
|
require.Len(t, remote.archiveRequests, 2)
|
|
assert.Equal(t, []string{changed}, remote.archiveRequests[1].DeltaFiles)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestPreparedHTTPSyncContributorKeepsLockUntilClose(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
remote.writeSession(t, "session.jsonl",
|
|
time.Date(2026, 7, 8, 10, 0, 0, 0, time.UTC), "locked contributor")
|
|
database, hs := newMirrorSync(t, remote, t.TempDir())
|
|
prepared, err := hs.Prepare(context.Background())
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { require.NoError(t, prepared.Close()) })
|
|
contributor, err := prepared.RebuildContributor()
|
|
require.NoError(t, err)
|
|
|
|
contributorStarted := make(chan struct{})
|
|
continueContributor := make(chan struct{})
|
|
originalProgress := contributor.Progress
|
|
contributor.Progress = func(progress syncpkg.Progress) syncpkg.Progress {
|
|
if progress.Phase == syncpkg.PhaseDiscovering {
|
|
select {
|
|
case <-contributorStarted:
|
|
default:
|
|
close(contributorStarted)
|
|
<-continueContributor
|
|
}
|
|
}
|
|
return originalProgress(progress)
|
|
}
|
|
engine := syncpkg.NewEngine(database, syncpkg.EngineConfig{})
|
|
t.Cleanup(engine.Close)
|
|
type rebuildResult struct {
|
|
stats syncpkg.SyncStats
|
|
err error
|
|
}
|
|
rebuilt := make(chan rebuildResult, 1)
|
|
go func() {
|
|
stats, runErr := engine.ResyncAllWithOptions(
|
|
context.Background(), nil,
|
|
syncpkg.RebuildOptions{Contributors: []syncpkg.RebuildContributor{
|
|
contributor,
|
|
}},
|
|
)
|
|
rebuilt <- rebuildResult{stats: stats, err: runErr}
|
|
}()
|
|
select {
|
|
case <-contributorStarted:
|
|
case <-time.After(time.Second):
|
|
require.FailNow(t, "timed out waiting for contributor")
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond)
|
|
competing, lockErr := AcquireMirrorLock(ctx, prepared.Root())
|
|
cancel()
|
|
if competing != nil {
|
|
require.NoError(t, competing.Close())
|
|
}
|
|
assert.Error(t, lockErr, "prepared source must retain its lock during contribution")
|
|
close(continueContributor)
|
|
select {
|
|
case result := <-rebuilt:
|
|
require.NoError(t, result.err)
|
|
assert.False(t, result.stats.Aborted)
|
|
case <-time.After(time.Second):
|
|
require.FailNow(t, "timed out waiting for rebuild")
|
|
}
|
|
|
|
require.NoError(t, prepared.Close())
|
|
assertMirrorUnlocked(t, MirrorDir(hs.DataDir, hs.Host))
|
|
}
|
|
|
|
func TestPrepareHTTPSyncUnchangedFullMakesNoArchiveRequest(t *testing.T) {
|
|
for _, fileCount := range []int{5, 500} {
|
|
t.Run(strconv.Itoa(fileCount), func(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
mtime := time.Date(2026, 7, 8, 10, 0, 0, 0, time.UTC)
|
|
for i := range fileCount {
|
|
remote.writeSession(t, fmt.Sprintf("%03d.jsonl", i), mtime,
|
|
fmt.Sprintf("session %d", i))
|
|
}
|
|
_, hs := newMirrorSync(t, remote, t.TempDir())
|
|
|
|
prepared, err := hs.Prepare(context.Background())
|
|
require.NoError(t, err)
|
|
require.NoError(t, prepared.Close())
|
|
require.Len(t, remote.archiveRequests, 1)
|
|
|
|
hs.Full = true
|
|
prepared, err = hs.Prepare(context.Background())
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { require.NoError(t, prepared.Close()) })
|
|
assert.Len(t, remote.archiveRequests, 1,
|
|
"unchanged full preparation must not transfer the %d-file mirror",
|
|
fileCount)
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestPrepareHTTPSyncManualMirrorRemovalBootstrapsFull(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
remote.writeSession(t, "a.jsonl",
|
|
time.Date(2026, 7, 8, 10, 0, 0, 0, time.UTC), "session a")
|
|
dataDir := t.TempDir()
|
|
_, hs := newMirrorSync(t, remote, dataDir)
|
|
|
|
prepared, err := hs.Prepare(context.Background())
|
|
require.NoError(t, err)
|
|
require.NoError(t, prepared.Close())
|
|
require.Len(t, remote.archiveRequests, 1)
|
|
|
|
require.NoError(t, os.RemoveAll(MirrorDir(dataDir, "devbox")))
|
|
prepared, err = hs.Prepare(context.Background())
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { require.NoError(t, prepared.Close()) })
|
|
|
|
require.Len(t, remote.archiveRequests, 2)
|
|
assert.Nil(t, remote.archiveRequests[1].DeltaFiles)
|
|
}
|
|
|
|
func TestPreparedHTTPSyncCloseReleasesLockAndRemovesLegacyRoot(t *testing.T) {
|
|
t.Run("manifest mirror persists", func(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
remote.writeSession(t, "a.jsonl",
|
|
time.Date(2026, 7, 8, 10, 0, 0, 0, time.UTC), "session a")
|
|
dataDir := t.TempDir()
|
|
_, hs := newMirrorSync(t, remote, dataDir)
|
|
|
|
prepared, err := hs.Prepare(context.Background())
|
|
require.NoError(t, err)
|
|
root := prepared.Root()
|
|
assert.Equal(t, MirrorDir(dataDir, "devbox"), root)
|
|
assert.DirExists(t, root)
|
|
assert.Equal(t, remote.targets, prepared.Targets())
|
|
assertMirrorLocked(t, root)
|
|
|
|
require.NoError(t, prepared.Close())
|
|
require.NoError(t, prepared.Close(), "persistent Close must be idempotent")
|
|
assert.DirExists(t, root, "persistent mirror must survive Close")
|
|
assertMirrorUnlocked(t, root)
|
|
})
|
|
|
|
t.Run("legacy root is owned", func(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
remote.manifestStatus = http.StatusNotFound
|
|
remote.writeSession(t, "a.jsonl",
|
|
time.Date(2026, 7, 8, 10, 0, 0, 0, time.UTC), "session a")
|
|
dataDir := t.TempDir()
|
|
_, hs := newMirrorSync(t, remote, dataDir)
|
|
|
|
prepared, err := hs.Prepare(context.Background())
|
|
require.NoError(t, err)
|
|
root := prepared.Root()
|
|
assert.DirExists(t, root)
|
|
assert.NotEqual(t, MirrorDir(dataDir, "devbox"), root)
|
|
assert.NoDirExists(t, MirrorDir(dataDir, "devbox"))
|
|
assertMirrorLocked(t, MirrorDir(dataDir, "devbox"))
|
|
|
|
require.NoError(t, prepared.Close())
|
|
require.NoError(t, prepared.Close(), "legacy Close must be idempotent")
|
|
assert.NoDirExists(t, root, "owned legacy root must be removed")
|
|
assertMirrorUnlocked(t, MirrorDir(dataDir, "devbox"))
|
|
})
|
|
}
|
|
|
|
func TestPreparedHTTPSyncCloseRetriesFailedCleanup(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
remote.manifestStatus = http.StatusNotFound
|
|
remote.writeSession(t, "a.jsonl",
|
|
time.Date(2026, 7, 8, 10, 0, 0, 0, time.UTC), "session a")
|
|
dataDir := t.TempDir()
|
|
_, hs := newMirrorSync(t, remote, dataDir)
|
|
prepared, err := hs.Prepare(context.Background())
|
|
require.NoError(t, err)
|
|
root := prepared.Root()
|
|
|
|
removeErr := errors.New("transient remove failure")
|
|
unlockErr := errors.New("transient unlock failure")
|
|
removeCalls := 0
|
|
prepared.removeRoot = func(path string) error {
|
|
removeCalls++
|
|
if removeCalls == 1 {
|
|
return removeErr
|
|
}
|
|
return os.RemoveAll(path)
|
|
}
|
|
unlockCalls := 0
|
|
prepared.releaseLock = func(lock *MirrorLockHandle) error {
|
|
unlockCalls++
|
|
if unlockCalls == 1 {
|
|
return unlockErr
|
|
}
|
|
return lock.Close()
|
|
}
|
|
|
|
err = prepared.Close()
|
|
assert.ErrorIs(t, err, removeErr)
|
|
assert.ErrorIs(t, err, unlockErr)
|
|
assert.DirExists(t, root)
|
|
assertMirrorLocked(t, MirrorDir(dataDir, "devbox"))
|
|
|
|
require.NoError(t, prepared.Close())
|
|
assert.NoDirExists(t, root)
|
|
assertMirrorUnlocked(t, MirrorDir(dataDir, "devbox"))
|
|
require.NoError(t, prepared.Close(), "fully cleaned Close must be idempotent")
|
|
}
|
|
|
|
func TestPreparedHTTPSyncCloseTracksCleanupIndependently(t *testing.T) {
|
|
prepareLegacy := func(t *testing.T) (*PreparedHTTP, string) {
|
|
t.Helper()
|
|
remote := newMirrorTestRemote(t)
|
|
remote.manifestStatus = http.StatusNotFound
|
|
remote.writeSession(t, "a.jsonl",
|
|
time.Date(2026, 7, 8, 10, 0, 0, 0, time.UTC), "session a")
|
|
dataDir := t.TempDir()
|
|
_, hs := newMirrorSync(t, remote, dataDir)
|
|
prepared, err := hs.Prepare(context.Background())
|
|
require.NoError(t, err)
|
|
return prepared, MirrorDir(dataDir, "devbox")
|
|
}
|
|
|
|
t.Run("successful unlock is not repeated", func(t *testing.T) {
|
|
prepared, mirrorRoot := prepareLegacy(t)
|
|
removeErr := errors.New("transient remove failure")
|
|
removeCalls := 0
|
|
prepared.removeRoot = func(path string) error {
|
|
removeCalls++
|
|
if removeCalls == 1 {
|
|
return removeErr
|
|
}
|
|
return os.RemoveAll(path)
|
|
}
|
|
unlockCalls := 0
|
|
prepared.releaseLock = func(lock *MirrorLockHandle) error {
|
|
unlockCalls++
|
|
return lock.Close()
|
|
}
|
|
|
|
assert.ErrorIs(t, prepared.Close(), removeErr)
|
|
assertMirrorUnlocked(t, mirrorRoot)
|
|
require.NoError(t, prepared.Close())
|
|
assert.Equal(t, 1, unlockCalls)
|
|
})
|
|
|
|
t.Run("successful removal is not repeated", func(t *testing.T) {
|
|
prepared, mirrorRoot := prepareLegacy(t)
|
|
unlockErr := errors.New("transient unlock failure")
|
|
removeCalls := 0
|
|
prepared.removeRoot = func(path string) error {
|
|
removeCalls++
|
|
return os.RemoveAll(path)
|
|
}
|
|
unlockCalls := 0
|
|
prepared.releaseLock = func(lock *MirrorLockHandle) error {
|
|
unlockCalls++
|
|
if unlockCalls == 1 {
|
|
return unlockErr
|
|
}
|
|
return lock.Close()
|
|
}
|
|
|
|
assert.ErrorIs(t, prepared.Close(), unlockErr)
|
|
assert.NoDirExists(t, prepared.Root())
|
|
assertMirrorLocked(t, mirrorRoot)
|
|
require.NoError(t, prepared.Close())
|
|
assert.Equal(t, 1, removeCalls)
|
|
assertMirrorUnlocked(t, mirrorRoot)
|
|
})
|
|
}
|
|
|
|
func TestPrepareHTTPSyncClearsOnlySelectedHostCacheBeforeMirrorMutation(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
base := time.Date(2026, 7, 8, 10, 0, 0, 0, time.UTC)
|
|
changed := remote.writeSession(t, "a.jsonl", base, "session a")
|
|
remote.writeSession(t, "b.jsonl", base, "session b")
|
|
remote.writeSession(t, "c.jsonl", base, "session c")
|
|
database, hs := newMirrorSync(t, remote, t.TempDir())
|
|
prepared, err := hs.Prepare(context.Background())
|
|
require.NoError(t, err)
|
|
require.NoError(t, prepared.Close())
|
|
|
|
require.NoError(t, database.ReplaceRemoteSkippedFiles(
|
|
"devbox", map[string]int64{changed: 101},
|
|
))
|
|
require.NoError(t, database.ReplaceRemoteSkippedFiles(
|
|
"other-host", map[string]int64{"/remote/other.jsonl": 202},
|
|
))
|
|
remote.writeSession(t, "a.jsonl", base.Add(time.Second),
|
|
"session a", "continued")
|
|
type cacheSnapshot struct {
|
|
selected map[string]int64
|
|
other map[string]int64
|
|
err error
|
|
}
|
|
atArchive := make(chan cacheSnapshot, 1)
|
|
remote.onArchive = func(ArchiveRequest) {
|
|
selected, err := database.LoadRemoteSkippedFiles("devbox")
|
|
if err != nil {
|
|
atArchive <- cacheSnapshot{err: err}
|
|
return
|
|
}
|
|
other, err := database.LoadRemoteSkippedFiles("other-host")
|
|
atArchive <- cacheSnapshot{selected: selected, other: other, err: err}
|
|
}
|
|
|
|
prepared, err = hs.Prepare(context.Background())
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { require.NoError(t, prepared.Close()) })
|
|
var snapshot cacheSnapshot
|
|
select {
|
|
case snapshot = <-atArchive:
|
|
case <-time.After(time.Second):
|
|
require.FailNow(t, "timed out waiting for archive cache snapshot")
|
|
}
|
|
require.NoError(t, snapshot.err)
|
|
assert.Empty(t, snapshot.selected,
|
|
"selected cache must be clear before archive extraction starts")
|
|
assert.Equal(t, map[string]int64{"/remote/other.jsonl": 202}, snapshot.other)
|
|
selected, err := database.LoadRemoteSkippedFiles("devbox")
|
|
require.NoError(t, err)
|
|
assert.Empty(t, selected)
|
|
}
|
|
|
|
func TestPrepareHTTPSyncUnchangedPreservesSelectedHostCache(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
path := remote.writeSession(t, "a.jsonl",
|
|
time.Date(2026, 7, 8, 10, 0, 0, 0, time.UTC), "session a")
|
|
database, hs := newMirrorSync(t, remote, t.TempDir())
|
|
prepared, err := hs.Prepare(context.Background())
|
|
require.NoError(t, err)
|
|
require.NoError(t, prepared.Close())
|
|
|
|
require.NoError(t, database.ReplaceRemoteSkippedFiles(
|
|
"devbox", map[string]int64{path: 303},
|
|
))
|
|
require.NoError(t, database.ReplaceRemoteSkippedFiles(
|
|
"other-host", map[string]int64{"/remote/other.jsonl": 304},
|
|
))
|
|
requests := len(remote.archiveRequests)
|
|
prepared, err = hs.Prepare(context.Background())
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { require.NoError(t, prepared.Close()) })
|
|
assert.Len(t, remote.archiveRequests, requests)
|
|
selected, err := database.LoadRemoteSkippedFiles("devbox")
|
|
require.NoError(t, err)
|
|
assert.Equal(t, map[string]int64{path: 303}, selected)
|
|
other, err := database.LoadRemoteSkippedFiles("other-host")
|
|
require.NoError(t, err)
|
|
assert.Equal(t, map[string]int64{"/remote/other.jsonl": 304}, other)
|
|
}
|
|
|
|
func TestPrepareHTTPSyncFailureCleansOwnedResources(t *testing.T) {
|
|
t.Run("manifest failure releases lock", func(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
remote.manifestStatus = http.StatusInternalServerError
|
|
_, hs := newMirrorSync(t, remote, t.TempDir())
|
|
|
|
prepared, err := hs.Prepare(context.Background())
|
|
require.Error(t, err)
|
|
assert.Nil(t, prepared)
|
|
assertMirrorUnlocked(t, MirrorDir(hs.DataDir, "devbox"))
|
|
})
|
|
|
|
t.Run("legacy extraction failure removes temp root", func(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
remote.manifestStatus = http.StatusNotFound
|
|
path := remote.writeSession(t, "a.jsonl",
|
|
time.Date(2026, 7, 8, 10, 0, 0, 0, time.UTC), "session a")
|
|
name, err := safeRemotePathArchiveName(path)
|
|
require.NoError(t, err)
|
|
remote.archiveBody = tarWithoutEndMarker(t, name, "partial legacy")
|
|
tempParent := t.TempDir()
|
|
setPortableTempDir(t, tempParent)
|
|
_, hs := newMirrorSync(t, remote, t.TempDir())
|
|
|
|
prepared, err := hs.Prepare(context.Background())
|
|
require.Error(t, err)
|
|
assert.Nil(t, prepared)
|
|
roots, globErr := filepath.Glob(filepath.Join(tempParent, "agentsview-http-*"))
|
|
require.NoError(t, globErr)
|
|
assert.Empty(t, roots)
|
|
assert.NoDirExists(t, MirrorDir(hs.DataDir, "devbox"))
|
|
assertMirrorUnlocked(t, MirrorDir(hs.DataDir, "devbox"))
|
|
})
|
|
|
|
t.Run("failed mirror extraction leaves bytes and invalidated cache", func(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
path := remote.writeSession(t, "a.jsonl",
|
|
time.Date(2026, 7, 8, 10, 0, 0, 0, time.UTC), "session a")
|
|
name, err := safeRemotePathArchiveName(path)
|
|
require.NoError(t, err)
|
|
remote.archiveBody = tarWithoutEndMarker(t, name, "partial mirror")
|
|
database, hs := newMirrorSync(t, remote, t.TempDir())
|
|
require.NoError(t, database.ReplaceRemoteSkippedFiles(
|
|
"devbox", map[string]int64{path: 404},
|
|
))
|
|
require.NoError(t, database.ReplaceRemoteSkippedFiles(
|
|
"other-host", map[string]int64{"/remote/other.jsonl": 505},
|
|
))
|
|
|
|
prepared, err := hs.Prepare(context.Background())
|
|
require.Error(t, err)
|
|
assert.Nil(t, prepared)
|
|
local, mapErr := safeRemappedRemotePath(MirrorDir(hs.DataDir, "devbox"), path)
|
|
require.NoError(t, mapErr)
|
|
body, readErr := os.ReadFile(local)
|
|
require.NoError(t, readErr)
|
|
assert.Equal(t, "partial mirror", string(body))
|
|
selected, loadErr := database.LoadRemoteSkippedFiles("devbox")
|
|
require.NoError(t, loadErr)
|
|
assert.Empty(t, selected)
|
|
other, loadErr := database.LoadRemoteSkippedFiles("other-host")
|
|
require.NoError(t, loadErr)
|
|
assert.Equal(t, map[string]int64{"/remote/other.jsonl": 505}, other)
|
|
assertMirrorUnlocked(t, MirrorDir(hs.DataDir, "devbox"))
|
|
})
|
|
}
|
|
|
|
func TestPrepareHTTPSyncsSortsHostsAndUnwindsOnFailure(t *testing.T) {
|
|
remoteA := newMirrorTestRemote(t)
|
|
remoteA.manifestStatus = http.StatusNotFound
|
|
remoteA.writeSession(t, "a.jsonl",
|
|
time.Date(2026, 7, 11, 10, 0, 0, 0, time.UTC), "host a")
|
|
remoteB := newMirrorTestRemote(t)
|
|
remoteB.manifestStatus = http.StatusInternalServerError
|
|
remoteB.writeSession(t, "b.jsonl",
|
|
time.Date(2026, 7, 11, 10, 0, 0, 0, time.UTC), "host b")
|
|
|
|
dataDir := t.TempDir()
|
|
tempParent := t.TempDir()
|
|
setPortableTempDir(t, tempParent)
|
|
database, syncA := newMirrorSync(t, remoteA, dataDir)
|
|
syncA.Host = "host-a"
|
|
syncB := syncA
|
|
syncB.Host = "host-b"
|
|
syncB.URL = remoteB.ts.URL
|
|
syncB.DB = database
|
|
var requestOrder []string
|
|
remoteA.onManifest = func() { requestOrder = append(requestOrder, "host-a") }
|
|
remoteB.onManifest = func() { requestOrder = append(requestOrder, "host-b") }
|
|
|
|
syncs := []HTTPSync{syncB, syncA}
|
|
prepared, err := PrepareHTTPSyncs(context.Background(), syncs)
|
|
require.Error(t, err)
|
|
assert.Nil(t, prepared)
|
|
assert.Equal(t, []string{"host-b", "host-a"}, []string{
|
|
syncs[0].Host, syncs[1].Host,
|
|
}, "preparation must not mutate caller ordering")
|
|
assert.Equal(t, []string{"host-a", "host-b"}, requestOrder)
|
|
var hostErr *HostError
|
|
require.ErrorAs(t, err, &hostErr)
|
|
assert.Equal(t, "host-b", hostErr.Host)
|
|
assertMirrorUnlocked(t, MirrorDir(dataDir, "host-a"))
|
|
assertMirrorUnlocked(t, MirrorDir(dataDir, "host-b"))
|
|
roots, globErr := filepath.Glob(filepath.Join(tempParent, "agentsview-http-*"))
|
|
require.NoError(t, globErr)
|
|
assert.Empty(t, roots, "unwind removes the successful legacy root and spools")
|
|
}
|
|
|
|
func TestPrepareHTTPSyncsLegacySourcesDoNotCreateWorkingDirectoryLocks(t *testing.T) {
|
|
remoteA := newMirrorTestRemote(t)
|
|
remoteA.writeSession(t, "a.jsonl",
|
|
time.Date(2026, 7, 11, 11, 0, 0, 0, time.UTC), "host a")
|
|
remoteB := newMirrorTestRemote(t)
|
|
remoteB.writeSession(t, "b.jsonl",
|
|
time.Date(2026, 7, 11, 11, 0, 0, 0, time.UTC), "host b")
|
|
|
|
workingDir := t.TempDir()
|
|
previousDir, err := os.Getwd()
|
|
require.NoError(t, err)
|
|
require.NoError(t, os.Chdir(workingDir))
|
|
t.Cleanup(func() { require.NoError(t, os.Chdir(previousDir)) })
|
|
setPortableTempDir(t, t.TempDir())
|
|
|
|
var requestOrder []string
|
|
remoteA.onArchive = func(ArchiveRequest) {
|
|
requestOrder = append(requestOrder, "host-a")
|
|
}
|
|
remoteB.onArchive = func(ArchiveRequest) {
|
|
requestOrder = append(requestOrder, "host-b")
|
|
}
|
|
syncA := HTTPSync{Host: "host-a", URL: remoteA.ts.URL, Client: remoteA.ts.Client()}
|
|
syncB := HTTPSync{Host: "host-b", URL: remoteB.ts.URL, Client: remoteB.ts.Client()}
|
|
syncs := []HTTPSync{syncB, syncA}
|
|
|
|
prepared, err := PrepareHTTPSyncs(context.Background(), syncs)
|
|
require.NoError(t, err)
|
|
require.NotNil(t, prepared)
|
|
require.NoError(t, prepared.Close())
|
|
|
|
assert.Equal(t, []string{"host-b", "host-a"}, []string{
|
|
syncs[0].Host, syncs[1].Host,
|
|
}, "preparation must not mutate caller ordering")
|
|
assert.Equal(t, []string{"host-a", "host-b"}, requestOrder)
|
|
assert.NoDirExists(t, filepath.Join(workingDir, "remote-mirrors"),
|
|
"legacy sources have no persistent mirror lock identity")
|
|
}
|
|
|
|
func TestPrepareHTTPSyncsOrdersConcurrentCallersByMirrorLockPath(t *testing.T) {
|
|
remoteA := newMirrorTestRemote(t)
|
|
remoteA.writeSession(t, "a.jsonl",
|
|
time.Date(2026, 7, 11, 14, 0, 0, 0, time.UTC), "mirror a")
|
|
remoteB := newMirrorTestRemote(t)
|
|
remoteB.writeSession(t, "b.jsonl",
|
|
time.Date(2026, 7, 11, 14, 0, 0, 0, time.UTC), "mirror b")
|
|
|
|
base := t.TempDir()
|
|
dataDirA := base + string(filepath.Separator) + "z" +
|
|
string(filepath.Separator) + ".." + string(filepath.Separator) + "a-data"
|
|
dataDirB := filepath.Join(base, "b-data")
|
|
database, syncA := newMirrorSync(t, remoteA, dataDirA)
|
|
syncA.Host = "shared-host"
|
|
syncB := syncA
|
|
syncB.DataDir = dataDirB
|
|
syncB.URL = remoteB.ts.URL
|
|
syncB.DB = database
|
|
assert.Less(t, MirrorDir(dataDirA, syncA.Host), MirrorDir(dataDirB, syncB.Host))
|
|
|
|
firstA := make(chan struct{})
|
|
firstB := make(chan struct{})
|
|
var onceA, onceB stdsync.Once
|
|
var orderMu stdsync.Mutex
|
|
orders := map[string][]string{}
|
|
record := func(
|
|
caller, mirror string, ownOnce *stdsync.Once,
|
|
own chan struct{}, other <-chan struct{},
|
|
) {
|
|
orderMu.Lock()
|
|
orders[caller] = append(orders[caller], mirror)
|
|
orderMu.Unlock()
|
|
ownOnce.Do(func() { close(own) })
|
|
select {
|
|
case <-other:
|
|
case <-time.After(50 * time.Millisecond):
|
|
}
|
|
}
|
|
remoteA.onManifestRequest = func(r *http.Request) {
|
|
record(r.Header.Get("Authorization"), "mirror-a",
|
|
&onceA, firstA, firstB)
|
|
}
|
|
remoteB.onManifestRequest = func(r *http.Request) {
|
|
record(r.Header.Get("Authorization"), "mirror-b",
|
|
&onceB, firstB, firstA)
|
|
}
|
|
|
|
callerOne := []HTTPSync{syncB, syncA}
|
|
callerOne[0].Token = "caller-one"
|
|
callerOne[1].Token = "caller-one"
|
|
callerTwo := []HTTPSync{syncA, syncB}
|
|
callerTwo[0].Token = "caller-two"
|
|
callerTwo[1].Token = "caller-two"
|
|
type result struct {
|
|
caller string
|
|
prepared *PreparedHTTPSyncs
|
|
err error
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
|
|
defer cancel()
|
|
start := make(chan struct{})
|
|
results := make(chan result, 2)
|
|
for caller, syncs := range map[string][]HTTPSync{
|
|
"caller-one": callerOne,
|
|
"caller-two": callerTwo,
|
|
} {
|
|
go func() {
|
|
<-start
|
|
prepared, err := PrepareHTTPSyncs(ctx, syncs)
|
|
results <- result{caller: caller, prepared: prepared, err: err}
|
|
}()
|
|
}
|
|
close(start)
|
|
|
|
got := make(map[string]error, 2)
|
|
for range 2 {
|
|
select {
|
|
case result := <-results:
|
|
got[result.caller] = result.err
|
|
if result.prepared != nil {
|
|
require.NoError(t, result.prepared.Close())
|
|
}
|
|
case <-ctx.Done():
|
|
require.FailNow(t, "concurrent preparation deadlocked",
|
|
"orders=%v", orders)
|
|
}
|
|
}
|
|
require.NoError(t, got["caller-one"])
|
|
require.NoError(t, got["caller-two"])
|
|
orderMu.Lock()
|
|
assert.Equal(t, []string{"mirror-a", "mirror-b"}, orders["Bearer caller-one"])
|
|
assert.Equal(t, []string{"mirror-a", "mirror-b"}, orders["Bearer caller-two"])
|
|
orderMu.Unlock()
|
|
assert.Equal(t, []string{"shared-host", "shared-host"}, []string{
|
|
callerOne[0].Host, callerOne[1].Host,
|
|
})
|
|
assert.Equal(t, []string{dataDirB, dataDirA}, []string{
|
|
callerOne[0].DataDir, callerOne[1].DataDir,
|
|
}, "preparation must preserve caller-one input order")
|
|
assert.Equal(t, []string{dataDirA, dataDirB}, []string{
|
|
callerTwo[0].DataDir, callerTwo[1].DataDir,
|
|
}, "preparation must preserve caller-two input order")
|
|
}
|
|
|
|
func TestPrepareHTTPSyncsRejectsDuplicateCanonicalLockIdentity(t *testing.T) {
|
|
t.Run("clean parent alias", func(t *testing.T) {
|
|
base := t.TempDir()
|
|
canonical := filepath.Join(base, "data")
|
|
alias := base + string(filepath.Separator) + "z" +
|
|
string(filepath.Separator) + ".." + string(filepath.Separator) + "data"
|
|
assertDuplicateCanonicalLockIdentity(t, canonical, alias)
|
|
})
|
|
|
|
t.Run("symlink parent alias", func(t *testing.T) {
|
|
base := t.TempDir()
|
|
realParent := filepath.Join(base, "real")
|
|
require.NoError(t, os.MkdirAll(realParent, 0o755))
|
|
linkParent := filepath.Join(base, "link")
|
|
if err := os.Symlink(realParent, linkParent); err != nil {
|
|
t.Skipf("symlinks unavailable: %v", err)
|
|
}
|
|
assertDuplicateCanonicalLockIdentity(
|
|
t, filepath.Join(realParent, "data"), filepath.Join(linkParent, "data"),
|
|
)
|
|
})
|
|
|
|
t.Run("case alias on insensitive volume", func(t *testing.T) {
|
|
base := t.TempDir()
|
|
probe := filepath.Join(base, "CaseSensitiveProbe")
|
|
require.NoError(t, os.Mkdir(probe, 0o755))
|
|
probeAlias := filepath.Join(base, "cASEsENSITIVEpROBE")
|
|
actualInfo, err := os.Stat(probe)
|
|
require.NoError(t, err)
|
|
aliasInfo, err := os.Stat(probeAlias)
|
|
if os.IsNotExist(err) {
|
|
t.Skip("test volume is case-sensitive")
|
|
}
|
|
require.NoError(t, err)
|
|
if !os.SameFile(actualInfo, aliasInfo) {
|
|
t.Skip("case aliases do not identify the same directory")
|
|
}
|
|
require.NoError(t, os.Remove(probe))
|
|
actualParent := filepath.Join(base, "InitiallyMissingCaseParent")
|
|
aliasParent := filepath.Join(base, "iNITIALLYmISSINGcASEpARENT")
|
|
assert.NoDirExists(t, actualParent)
|
|
assert.NoDirExists(t, aliasParent)
|
|
assertDuplicateCanonicalLockIdentity(
|
|
t, filepath.Join(actualParent, "data"), filepath.Join(aliasParent, "data"),
|
|
)
|
|
})
|
|
|
|
t.Run("unicode normalization alias", func(t *testing.T) {
|
|
base := t.TempDir()
|
|
probe := filepath.Join(base, "Caf\u00e9Probe")
|
|
probeAlias := filepath.Join(base, "Cafe\u0301Probe")
|
|
require.NoError(t, os.Mkdir(probe, 0o755))
|
|
actualInfo, err := os.Stat(probe)
|
|
require.NoError(t, err)
|
|
aliasInfo, err := os.Stat(probeAlias)
|
|
if os.IsNotExist(err) {
|
|
t.Skip("test volume does not alias Unicode normalization forms")
|
|
}
|
|
require.NoError(t, err)
|
|
if !os.SameFile(actualInfo, aliasInfo) {
|
|
t.Skip("Unicode normalization forms identify different directories")
|
|
}
|
|
require.NoError(t, os.Remove(probe))
|
|
actualParent := filepath.Join(base, "Caf\u00e9Missing")
|
|
aliasParent := filepath.Join(base, "Cafe\u0301Missing")
|
|
assert.NoDirExists(t, actualParent)
|
|
assert.NoDirExists(t, aliasParent)
|
|
assertDuplicateCanonicalLockIdentity(
|
|
t, filepath.Join(actualParent, "data"), filepath.Join(aliasParent, "data"),
|
|
)
|
|
})
|
|
}
|
|
|
|
func assertDuplicateCanonicalLockIdentity(t *testing.T, dataDirA, dataDirB string) {
|
|
t.Helper()
|
|
remote := newMirrorTestRemote(t)
|
|
remote.writeSession(t, "duplicate.jsonl",
|
|
time.Date(2026, 7, 11, 14, 30, 0, 0, time.UTC), "duplicate")
|
|
database, syncA := newMirrorSync(t, remote, dataDirA)
|
|
syncA.Host = "shared-host"
|
|
syncB := syncA
|
|
syncB.DataDir = dataDirB
|
|
syncB.DB = database
|
|
manifestRequests := 0
|
|
remote.onManifest = func() { manifestRequests++ }
|
|
syncs := []HTTPSync{syncB, syncA}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 750*time.Millisecond)
|
|
defer cancel()
|
|
prepared, err := PrepareHTTPSyncs(ctx, syncs)
|
|
require.Error(t, err)
|
|
assert.Nil(t, prepared)
|
|
assert.ErrorIs(t, err, ErrDuplicateMirrorLock)
|
|
var hostErr *HostError
|
|
require.ErrorAs(t, err, &hostErr)
|
|
assert.Equal(t, "shared-host", hostErr.Host)
|
|
assert.Equal(t, "resolve mirror lock", hostErr.Operation)
|
|
assert.Contains(t, hostErr.Error(), `"shared-host"`)
|
|
assert.Zero(t, manifestRequests,
|
|
"duplicates must fail before acquiring a lock or preparing a source")
|
|
lockArtifact := MirrorDir(dataDirA, syncA.Host) + ".lock"
|
|
lockInfo, statErr := os.Stat(lockArtifact)
|
|
require.NoError(t, statErr)
|
|
assert.False(t, lockInfo.IsDir())
|
|
if runtime.GOOS != "windows" {
|
|
assert.Equal(t, os.FileMode(0o600), lockInfo.Mode().Perm(),
|
|
"preflight creates only the same private lock artifact acquisition needs")
|
|
}
|
|
assert.Equal(t, []string{dataDirB, dataDirA}, []string{
|
|
syncs[0].DataDir, syncs[1].DataDir,
|
|
}, "duplicate detection must preserve caller order")
|
|
lockCtx, lockCancel := context.WithTimeout(context.Background(), time.Second)
|
|
defer lockCancel()
|
|
lock, lockErr := AcquireMirrorLock(lockCtx, MirrorDir(dataDirA, syncA.Host))
|
|
require.NoError(t, lockErr)
|
|
require.NoError(t, lock.Close())
|
|
}
|
|
|
|
func TestPreparedHTTPSyncsHoldAllLocksUntilClose(t *testing.T) {
|
|
remoteA := newMirrorTestRemote(t)
|
|
remoteA.writeSession(t, "a.jsonl",
|
|
time.Date(2026, 7, 11, 11, 0, 0, 0, time.UTC), "host a")
|
|
remoteB := newMirrorTestRemote(t)
|
|
remoteB.writeSession(t, "b.jsonl",
|
|
time.Date(2026, 7, 11, 11, 0, 0, 0, time.UTC), "host b")
|
|
dataDir := t.TempDir()
|
|
database, syncA := newMirrorSync(t, remoteA, dataDir)
|
|
syncA.Host = "host-a"
|
|
syncB := syncA
|
|
syncB.Host = "host-b"
|
|
syncB.URL = remoteB.ts.URL
|
|
syncB.DB = database
|
|
|
|
prepared, err := PrepareHTTPSyncs(
|
|
context.Background(), []HTTPSync{syncB, syncA},
|
|
)
|
|
require.NoError(t, err)
|
|
contributors, release, err := prepared.BorrowRebuildContributors()
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() {
|
|
release()
|
|
require.NoError(t, prepared.Close())
|
|
})
|
|
assert.Equal(t, []string{"host-a", "host-b"}, []string{
|
|
contributors[0].Name, contributors[1].Name,
|
|
})
|
|
|
|
for _, host := range []string{"host-a", "host-b"} {
|
|
lockCtx, cancel := context.WithTimeout(context.Background(), 40*time.Millisecond)
|
|
competing, lockErr := AcquireMirrorLock(
|
|
lockCtx, MirrorDir(dataDir, host),
|
|
)
|
|
cancel()
|
|
if competing != nil {
|
|
require.NoError(t, competing.Close())
|
|
}
|
|
assert.Error(t, lockErr,
|
|
"all mirror locks remain held across earlier contributor work")
|
|
}
|
|
|
|
release()
|
|
require.NoError(t, prepared.Close())
|
|
for _, host := range []string{"host-a", "host-b"} {
|
|
lockCtx, cancel := context.WithTimeout(context.Background(), time.Second)
|
|
competing, lockErr := AcquireMirrorLock(
|
|
lockCtx, MirrorDir(dataDir, host),
|
|
)
|
|
cancel()
|
|
require.NoError(t, lockErr, host)
|
|
require.NotNil(t, competing, host)
|
|
require.NoError(t, competing.Close())
|
|
}
|
|
_, _, err = prepared.BorrowRebuildContributors()
|
|
assert.ErrorIs(t, err, ErrPreparedClosed,
|
|
"closed aggregate cannot expose contributors")
|
|
}
|
|
|
|
func TestPreparedHTTPSyncsBorrowBlocksCloseUntilRelease(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
remote.writeSession(t, "borrow.jsonl",
|
|
time.Date(2026, 7, 11, 15, 30, 0, 0, time.UTC), "borrowed")
|
|
dataDir := t.TempDir()
|
|
_, hs := newMirrorSync(t, remote, dataDir)
|
|
prepared, err := PrepareHTTPSyncs(context.Background(), []HTTPSync{hs})
|
|
require.NoError(t, err)
|
|
contributors, release, err := prepared.BorrowRebuildContributors()
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() {
|
|
release()
|
|
require.NoError(t, prepared.Close())
|
|
})
|
|
require.Len(t, contributors, 1)
|
|
assert.Equal(t, hs.Host, contributors[0].Name)
|
|
|
|
assert.ErrorIs(t, prepared.Close(), ErrPreparedInUse)
|
|
_, _, err = prepared.BorrowRebuildContributors()
|
|
assert.ErrorIs(t, err, ErrPreparedClosed,
|
|
"a Close attempt ends new borrowing while retained ownership remains retryable")
|
|
lockCtx, cancel := context.WithTimeout(context.Background(), 40*time.Millisecond)
|
|
competing, lockErr := AcquireMirrorLock(lockCtx, MirrorDir(dataDir, hs.Host))
|
|
cancel()
|
|
if competing != nil {
|
|
require.NoError(t, competing.Close())
|
|
}
|
|
assert.Error(t, lockErr, "borrowed Close must retain its mirror lock")
|
|
|
|
release()
|
|
release()
|
|
require.NoError(t, prepared.Close())
|
|
lockCtx, cancel = context.WithTimeout(context.Background(), time.Second)
|
|
competing, lockErr = AcquireMirrorLock(lockCtx, MirrorDir(dataDir, hs.Host))
|
|
cancel()
|
|
require.NoError(t, lockErr)
|
|
require.NotNil(t, competing)
|
|
require.NoError(t, competing.Close())
|
|
_, _, err = prepared.BorrowRebuildContributors()
|
|
assert.ErrorIs(t, err, ErrPreparedClosed)
|
|
}
|
|
|
|
func TestPreparedHTTPSyncsBorrowAndCloseAreRaceSafe(t *testing.T) {
|
|
prepared := &PreparedHTTPSyncs{sources: []*PreparedHTTP{{
|
|
sync: HTTPSync{Host: "race-host"},
|
|
lock: &MirrorLockHandle{},
|
|
releaseLock: func(*MirrorLockHandle) error { return nil },
|
|
}}}
|
|
_, initialRelease, err := prepared.BorrowRebuildContributors()
|
|
require.NoError(t, err)
|
|
start := make(chan struct{})
|
|
borrowResults := make(chan error, 8)
|
|
closeResult := make(chan error, 1)
|
|
for range 8 {
|
|
go func() {
|
|
<-start
|
|
_, release, borrowErr := prepared.BorrowRebuildContributors()
|
|
if borrowErr == nil {
|
|
release()
|
|
release()
|
|
}
|
|
borrowResults <- borrowErr
|
|
}()
|
|
}
|
|
go func() {
|
|
<-start
|
|
closeResult <- prepared.Close()
|
|
}()
|
|
close(start)
|
|
for range 8 {
|
|
borrowErr := <-borrowResults
|
|
assert.True(t, borrowErr == nil || errors.Is(borrowErr, ErrPreparedClosed),
|
|
"unexpected borrow result: %v", borrowErr)
|
|
}
|
|
assert.ErrorIs(t, <-closeResult, ErrPreparedInUse)
|
|
initialRelease()
|
|
require.NoError(t, prepared.Close())
|
|
}
|
|
|
|
func TestPreparedHTTPSyncsCloseReversesJoinsAndRetries(t *testing.T) {
|
|
transient := errors.New("transient release")
|
|
var order []string
|
|
bCalls := 0
|
|
prepared := &PreparedHTTPSyncs{sources: []*PreparedHTTP{
|
|
{
|
|
lock: &MirrorLockHandle{},
|
|
releaseLock: func(*MirrorLockHandle) error {
|
|
order = append(order, "host-a")
|
|
return nil
|
|
},
|
|
},
|
|
{
|
|
lock: &MirrorLockHandle{},
|
|
releaseLock: func(*MirrorLockHandle) error {
|
|
order = append(order, "host-b")
|
|
bCalls++
|
|
if bCalls == 1 {
|
|
return transient
|
|
}
|
|
return nil
|
|
},
|
|
},
|
|
}}
|
|
|
|
assert.ErrorIs(t, prepared.Close(), transient)
|
|
assert.Equal(t, []string{"host-b", "host-a"}, order)
|
|
require.NoError(t, prepared.Close())
|
|
assert.Equal(t, []string{"host-b", "host-a", "host-b"}, order,
|
|
"only failed ownership is retained for retry")
|
|
require.NoError(t, prepared.Close(), "fully closed aggregate is idempotent")
|
|
}
|
|
|
|
func TestPrepareHTTPSyncsReturnsFailedUnwindOwnership(t *testing.T) {
|
|
remoteA := newMirrorTestRemote(t)
|
|
remoteA.manifestStatus = http.StatusNotFound
|
|
remoteA.writeSession(t, "a.jsonl",
|
|
time.Date(2026, 7, 11, 15, 0, 0, 0, time.UTC), "legacy a")
|
|
remoteB := newMirrorTestRemote(t)
|
|
remoteB.manifestStatus = http.StatusInternalServerError
|
|
remoteB.writeSession(t, "b.jsonl",
|
|
time.Date(2026, 7, 11, 15, 0, 0, 0, time.UTC), "host b")
|
|
tempParent := t.TempDir()
|
|
setPortableTempDir(t, tempParent)
|
|
dataDir := t.TempDir()
|
|
database, syncA := newMirrorSync(t, remoteA, dataDir)
|
|
syncA.Host = "host-a"
|
|
syncB := syncA
|
|
syncB.Host = "host-b"
|
|
syncB.URL = remoteB.ts.URL
|
|
syncB.DB = database
|
|
removeErr := errors.New("transient aggregate remove failure")
|
|
unlockErr := errors.New("transient aggregate unlock failure")
|
|
removeCalls := 0
|
|
unlockCalls := 0
|
|
var ownedRoot string
|
|
|
|
prepared, err := prepareHTTPSyncs(
|
|
context.Background(), []HTTPSync{syncB, syncA},
|
|
func(ctx context.Context, hs HTTPSync) (*PreparedHTTP, error) {
|
|
source, prepareErr := hs.Prepare(ctx)
|
|
if prepareErr != nil || hs.Host != "host-a" {
|
|
return source, prepareErr
|
|
}
|
|
ownedRoot = source.Root()
|
|
source.removeRoot = func(path string) error {
|
|
removeCalls++
|
|
if removeCalls == 1 {
|
|
return removeErr
|
|
}
|
|
return os.RemoveAll(path)
|
|
}
|
|
source.releaseLock = func(lock *MirrorLockHandle) error {
|
|
unlockCalls++
|
|
if unlockCalls == 1 {
|
|
return unlockErr
|
|
}
|
|
return lock.Close()
|
|
}
|
|
return source, nil
|
|
},
|
|
)
|
|
require.Error(t, err)
|
|
require.NotNil(t, prepared,
|
|
"failed cleanup ownership must be returned with the preparation error")
|
|
assert.ErrorIs(t, err, removeErr)
|
|
assert.ErrorIs(t, err, unlockErr)
|
|
var primary *HostError
|
|
require.ErrorAs(t, err, &primary)
|
|
assert.Equal(t, "host-b", primary.Host)
|
|
assert.Equal(t, "prepare", primary.Operation)
|
|
assert.Contains(t, err.Error(), `HTTP host "host-a" cleanup prepared source`)
|
|
assert.DirExists(t, ownedRoot)
|
|
assertMirrorLocked(t, MirrorDir(dataDir, "host-a"))
|
|
|
|
require.NoError(t, prepared.Close())
|
|
assert.NoDirExists(t, ownedRoot)
|
|
assertMirrorUnlocked(t, MirrorDir(dataDir, "host-a"))
|
|
require.NoError(t, prepared.Close(), "cleanup retry must remain idempotent")
|
|
}
|
|
|
|
func TestPrepareHTTPSyncRetainsCurrentSourceWhenCleanupFails(t *testing.T) {
|
|
t.Run("single source", func(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
remote.manifestStatus = http.StatusInternalServerError
|
|
_, hs := newMirrorSync(t, remote, t.TempDir())
|
|
unlockErr := errors.New("transient current-source unlock failure")
|
|
unlockCalls := 0
|
|
|
|
source, err := hs.prepare(context.Background(), func(source *PreparedHTTP) {
|
|
source.releaseLock = func(lock *MirrorLockHandle) error {
|
|
unlockCalls++
|
|
if unlockCalls == 1 {
|
|
return unlockErr
|
|
}
|
|
return lock.Close()
|
|
}
|
|
})
|
|
require.Error(t, err)
|
|
require.NotNil(t, source)
|
|
assert.ErrorIs(t, err, unlockErr)
|
|
var statusErr *StatusError
|
|
assert.ErrorAs(t, err, &statusErr)
|
|
assertMirrorLocked(t, MirrorDir(hs.DataDir, hs.Host))
|
|
|
|
require.NoError(t, source.Close())
|
|
assertMirrorUnlocked(t, MirrorDir(hs.DataDir, hs.Host))
|
|
})
|
|
|
|
t.Run("aggregate current source", func(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
remote.manifestStatus = http.StatusInternalServerError
|
|
_, hs := newMirrorSync(t, remote, t.TempDir())
|
|
unlockErr := errors.New("transient aggregate current-source unlock failure")
|
|
unlockCalls := 0
|
|
|
|
prepared, err := prepareHTTPSyncs(
|
|
context.Background(), []HTTPSync{hs},
|
|
func(ctx context.Context, hs HTTPSync) (*PreparedHTTP, error) {
|
|
return hs.prepare(ctx, func(source *PreparedHTTP) {
|
|
source.releaseLock = func(lock *MirrorLockHandle) error {
|
|
unlockCalls++
|
|
if unlockCalls <= 2 {
|
|
return unlockErr
|
|
}
|
|
return lock.Close()
|
|
}
|
|
})
|
|
},
|
|
)
|
|
require.Error(t, err)
|
|
require.NotNil(t, prepared)
|
|
assert.ErrorIs(t, err, unlockErr)
|
|
var hostErr *HostError
|
|
require.ErrorAs(t, err, &hostErr)
|
|
assert.Equal(t, hs.Host, hostErr.Host)
|
|
assert.Equal(t, "prepare", hostErr.Operation)
|
|
assert.Contains(t, err.Error(), "cleanup prepared source")
|
|
assertMirrorLocked(t, MirrorDir(hs.DataDir, hs.Host))
|
|
|
|
require.NoError(t, prepared.Close())
|
|
assertMirrorUnlocked(t, MirrorDir(hs.DataDir, hs.Host))
|
|
})
|
|
}
|
|
|
|
func TestHTTPSyncRunHandlesPreparationCleanupOwnership(t *testing.T) {
|
|
t.Run("transient cleanup is completed before return", func(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
remote.manifestStatus = http.StatusInternalServerError
|
|
_, hs := newMirrorSync(t, remote, t.TempDir())
|
|
unlockErr := errors.New("transient Run unlock failure")
|
|
unlockCalls := 0
|
|
base := hs
|
|
hs.runPrepare = func(ctx context.Context) (*PreparedHTTP, error) {
|
|
return base.prepare(ctx, func(source *PreparedHTTP) {
|
|
source.releaseLock = func(lock *MirrorLockHandle) error {
|
|
unlockCalls++
|
|
if unlockCalls == 1 {
|
|
return unlockErr
|
|
}
|
|
return lock.Close()
|
|
}
|
|
})
|
|
}
|
|
|
|
_, err := hs.Run(context.Background())
|
|
require.Error(t, err)
|
|
assert.ErrorIs(t, err, unlockErr)
|
|
var owner *PreparedCleanupError
|
|
assert.False(t, errors.As(err, &owner),
|
|
"successful Run retry must not return cleanup ownership")
|
|
assert.Equal(t, 2, unlockCalls)
|
|
assertMirrorUnlocked(t, MirrorDir(hs.DataDir, hs.Host))
|
|
})
|
|
|
|
t.Run("persistent cleanup ownership is retryable from error", func(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
remote.manifestStatus = http.StatusInternalServerError
|
|
_, hs := newMirrorSync(t, remote, t.TempDir())
|
|
unlockErr := errors.New("persistent Run unlock failure")
|
|
unlockCalls := 0
|
|
base := hs
|
|
hs.runPrepare = func(ctx context.Context) (*PreparedHTTP, error) {
|
|
return base.prepare(ctx, func(source *PreparedHTTP) {
|
|
source.releaseLock = func(lock *MirrorLockHandle) error {
|
|
unlockCalls++
|
|
if unlockCalls <= 2 {
|
|
return unlockErr
|
|
}
|
|
return lock.Close()
|
|
}
|
|
})
|
|
}
|
|
|
|
_, err := hs.Run(context.Background())
|
|
require.Error(t, err)
|
|
assert.ErrorIs(t, err, unlockErr)
|
|
var statusErr *StatusError
|
|
assert.ErrorAs(t, err, &statusErr)
|
|
var owner *PreparedCleanupError
|
|
require.ErrorAs(t, err, &owner)
|
|
assert.Equal(t, 2, unlockCalls, "Run must not double-close before returning")
|
|
assertMirrorLocked(t, MirrorDir(hs.DataDir, hs.Host))
|
|
|
|
require.NoError(t, owner.RetryCleanup())
|
|
assert.Equal(t, 3, unlockCalls)
|
|
assertMirrorUnlocked(t, MirrorDir(hs.DataDir, hs.Host))
|
|
require.NoError(t, owner.RetryCleanup(), "cleanup retry is idempotent")
|
|
assert.Equal(t, 3, unlockCalls)
|
|
})
|
|
}
|
|
|
|
func TestCleanupRegistryBlocksLaterRunWithPreparedCleanupError(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
remote.manifestStatus = http.StatusInternalServerError
|
|
_, hs := newMirrorSync(t, remote, t.TempDir())
|
|
unlockErr := errors.New("persistent registry unlock failure")
|
|
unlockCalls := 0
|
|
base := hs
|
|
hs.runPrepare = func(ctx context.Context) (*PreparedHTTP, error) {
|
|
return base.prepare(ctx, func(source *PreparedHTTP) {
|
|
source.releaseLock = func(lock *MirrorLockHandle) error {
|
|
unlockCalls++
|
|
if unlockCalls <= 4 {
|
|
return unlockErr
|
|
}
|
|
return lock.Close()
|
|
}
|
|
})
|
|
}
|
|
var registry CleanupRegistry
|
|
callbacks := 0
|
|
|
|
_, err := registry.Run(func() (SyncStats, error) {
|
|
callbacks++
|
|
return hs.Run(context.Background())
|
|
})
|
|
require.Error(t, err)
|
|
var owner *PreparedCleanupError
|
|
require.ErrorAs(t, err, &owner)
|
|
assert.ErrorIs(t, err, unlockErr)
|
|
assert.Equal(t, 3, unlockCalls)
|
|
assertMirrorLocked(t, MirrorDir(hs.DataDir, hs.Host))
|
|
|
|
_, err = registry.Run(func() (SyncStats, error) {
|
|
callbacks++
|
|
return SyncStats{}, nil
|
|
})
|
|
var pending *PendingCleanupError
|
|
require.ErrorAs(t, err, &pending)
|
|
var retained *PreparedCleanupError
|
|
require.ErrorAs(t, err, &retained)
|
|
assert.Same(t, owner, retained)
|
|
assert.ErrorIs(t, err, owner)
|
|
assert.ErrorIs(t, err, unlockErr)
|
|
assert.Equal(t, 1, callbacks, "pending cleanup must block the later callback")
|
|
assert.Equal(t, 4, unlockCalls)
|
|
assertMirrorLocked(t, MirrorDir(hs.DataDir, hs.Host))
|
|
|
|
_, err = registry.Run(func() (SyncStats, error) {
|
|
callbacks++
|
|
return SyncStats{SessionsSynced: 1}, nil
|
|
})
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 2, callbacks)
|
|
assert.Equal(t, 5, unlockCalls)
|
|
assertMirrorUnlocked(t, MirrorDir(hs.DataDir, hs.Host))
|
|
}
|
|
|
|
func TestPreparedHTTPCloseAttemptMakesSourceUnusableUntilCleanupCompletes(t *testing.T) {
|
|
unlockErr := errors.New("unlock remains pending")
|
|
unlockCalls := 0
|
|
prepared := &PreparedHTTP{
|
|
root: t.TempDir(),
|
|
lock: &MirrorLockHandle{},
|
|
releaseLock: func(*MirrorLockHandle) error {
|
|
unlockCalls++
|
|
if unlockCalls == 1 {
|
|
return unlockErr
|
|
}
|
|
return nil
|
|
},
|
|
}
|
|
|
|
require.ErrorIs(t, prepared.Close(), unlockErr)
|
|
_, importErr := prepared.ImportActive(context.Background())
|
|
assert.ErrorContains(t, importErr, "closed")
|
|
_, contributorErr := prepared.RebuildContributor()
|
|
assert.ErrorContains(t, contributorErr, "closed")
|
|
require.NoError(t, prepared.Close())
|
|
assert.Equal(t, 2, unlockCalls)
|
|
}
|
|
|
|
type failingArchiveBody struct{ err error }
|
|
|
|
func (r failingArchiveBody) Read([]byte) (int, error) { return 0, r.err }
|
|
func (failingArchiveBody) Close() error { return nil }
|
|
|
|
func TestCleanupRegistryRetainsFailedArchiveSpoolCleanup(t *testing.T) {
|
|
spoolRoot := t.TempDir()
|
|
readErr := errors.New("archive transfer failed")
|
|
removeErr := errors.New("spool removal failed")
|
|
removeCalls := 0
|
|
hs := HTTPSync{removeArchiveSpool: func(path string) error {
|
|
removeCalls++
|
|
if removeCalls <= 2 {
|
|
return removeErr
|
|
}
|
|
return os.RemoveAll(path)
|
|
}}
|
|
registry := new(CleanupRegistry)
|
|
callbacks := 0
|
|
|
|
_, err := registry.Run(func() (SyncStats, error) {
|
|
callbacks++
|
|
archive, downloadErr := hs.downloadArchive(
|
|
context.Background(),
|
|
&http.Response{
|
|
StatusCode: http.StatusOK,
|
|
Body: failingArchiveBody{err: readErr},
|
|
},
|
|
"Downloading archive", spoolRoot,
|
|
)
|
|
assert.Nil(t, archive)
|
|
return SyncStats{}, downloadErr
|
|
})
|
|
require.Error(t, err)
|
|
assert.ErrorIs(t, err, readErr)
|
|
assert.ErrorIs(t, err, removeErr)
|
|
var owner *downloadedArchiveCleanupError
|
|
require.ErrorAs(t, err, &owner)
|
|
assert.Equal(t, 2, removeCalls, "registry retries cleanup before retaining it")
|
|
|
|
_, err = registry.Run(func() (SyncStats, error) {
|
|
callbacks++
|
|
return SyncStats{SessionsSynced: 1}, nil
|
|
})
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 2, callbacks)
|
|
assert.Equal(t, 3, removeCalls)
|
|
entries, readDirErr := os.ReadDir(spoolRoot)
|
|
require.NoError(t, readDirErr)
|
|
assert.Empty(t, entries)
|
|
}
|
|
|
|
func TestPrepareHTTPSyncsCacheInvalidationFailureDoesNotMutateMirror(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
base := time.Date(2026, 7, 11, 12, 0, 0, 0, time.UTC)
|
|
path := remote.writeSession(t, "a.jsonl", base, "original")
|
|
dataDir := t.TempDir()
|
|
database, hs := newMirrorSync(t, remote, dataDir)
|
|
hs.Host = "readonly-host"
|
|
prepared, err := hs.Prepare(context.Background())
|
|
require.NoError(t, err)
|
|
require.NoError(t, prepared.Close())
|
|
|
|
local, err := safeRemappedRemotePath(MirrorDir(dataDir, hs.Host), path)
|
|
require.NoError(t, err)
|
|
beforeBytes, err := os.ReadFile(local)
|
|
require.NoError(t, err)
|
|
beforeInfo, err := os.Stat(local)
|
|
require.NoError(t, err)
|
|
requestsBefore := len(remote.archiveRequests)
|
|
remote.writeSession(t, "a.jsonl", base.Add(time.Second),
|
|
"replacement with a different size")
|
|
readonly, err := db.OpenReadOnly(database.Path())
|
|
require.NoError(t, err)
|
|
t.Cleanup(func() { require.NoError(t, readonly.Close()) })
|
|
hs.DB = readonly
|
|
|
|
preparedSet, err := PrepareHTTPSyncs(context.Background(), []HTTPSync{hs})
|
|
require.Error(t, err)
|
|
assert.Nil(t, preparedSet)
|
|
var hostErr *HostError
|
|
require.ErrorAs(t, err, &hostErr)
|
|
assert.Equal(t, "readonly-host", hostErr.Host)
|
|
assert.ErrorIs(t, err, db.ErrReadOnly)
|
|
afterBytes, readErr := os.ReadFile(local)
|
|
require.NoError(t, readErr)
|
|
afterInfo, statErr := os.Stat(local)
|
|
require.NoError(t, statErr)
|
|
assert.Equal(t, beforeBytes, afterBytes)
|
|
assert.Equal(t, beforeInfo.ModTime(), afterInfo.ModTime())
|
|
assert.Len(t, remote.archiveRequests, requestsBefore,
|
|
"cache failure must stop before archive download or extraction")
|
|
assertMirrorUnlocked(t, MirrorDir(dataDir, hs.Host))
|
|
}
|
|
|
|
func TestPrepareHTTPSyncSameMtimeSizeChangeRecoversAfterAbortedRebuild(t *testing.T) {
|
|
remote := newMirrorTestRemote(t)
|
|
base := time.Date(2026, 7, 11, 13, 0, 0, 0, time.UTC)
|
|
changed := remote.writeSession(t, "changed.jsonl", base,
|
|
"the original message is intentionally much longer")
|
|
deleted := remote.writeSession(t, "deleted.jsonl", base, "retained archive row")
|
|
dataDir := t.TempDir()
|
|
database, hs := newMirrorSync(t, remote, dataDir)
|
|
stats, err := hs.Run(context.Background())
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 2, stats.SessionsSynced)
|
|
initial, err := database.ListSessions(
|
|
context.Background(), db.SessionFilter{Limit: 10},
|
|
)
|
|
require.NoError(t, err)
|
|
var changedSessionID string
|
|
for _, session := range initial.Sessions {
|
|
messages, messageErr := database.GetMessages(
|
|
context.Background(), session.ID, 0, 10, true,
|
|
)
|
|
require.NoError(t, messageErr)
|
|
if len(messages) == 1 && strings.Contains(
|
|
messages[0].Content, "intentionally much longer",
|
|
) {
|
|
changedSessionID = session.ID
|
|
}
|
|
}
|
|
require.NotEmpty(t, changedSessionID)
|
|
|
|
require.NoError(t, database.ReplaceRemoteSkippedFiles(
|
|
hs.Host, map[string]int64{changed: base.UnixNano()},
|
|
))
|
|
remote.writeSession(t, "changed.jsonl", base, "new")
|
|
require.NoError(t, os.Remove(deleted))
|
|
prepared, err := hs.Prepare(context.Background())
|
|
require.NoError(t, err)
|
|
changedLocal, err := safeRemappedRemotePath(prepared.Root(), changed)
|
|
require.NoError(t, err)
|
|
changedBytes, err := os.ReadFile(changedLocal)
|
|
require.NoError(t, err)
|
|
assert.Contains(t, string(changedBytes), "new")
|
|
deletedLocal, err := safeRemappedRemotePath(prepared.Root(), deleted)
|
|
require.NoError(t, err)
|
|
assert.NoFileExists(t, deletedLocal)
|
|
cache, err := database.LoadRemoteSkippedFiles(hs.Host)
|
|
require.NoError(t, err)
|
|
assert.Empty(t, cache, "cache invalidation commits before mirror mutation")
|
|
require.NoError(t, prepared.Close(), "simulate rebuild abort before import")
|
|
|
|
stats, err = hs.Run(context.Background())
|
|
require.NoError(t, err)
|
|
assert.Equal(t, 1, stats.SessionsSynced)
|
|
page, err := database.ListSessions(context.Background(), db.SessionFilter{Limit: 10})
|
|
require.NoError(t, err)
|
|
require.Len(t, page.Sessions, 2,
|
|
"remote deletion must not delete the persistent archive row")
|
|
messages, err := database.GetMessages(
|
|
context.Background(), changedSessionID, 0, 10, true,
|
|
)
|
|
require.NoError(t, err)
|
|
require.Len(t, messages, 1)
|
|
assert.Equal(t, "new", messages[0].Content)
|
|
}
|
|
|
|
func tarWithoutEndMarker(t *testing.T, name, body string) []byte {
|
|
t.Helper()
|
|
archive := buildHTTPTestTar(t, map[string]string{name: body})
|
|
require.GreaterOrEqual(t, len(archive), tarEndMarkerSize)
|
|
return archive[:len(archive)-tarEndMarkerSize]
|
|
}
|
|
|
|
func setPortableTempDir(t *testing.T, dir string) {
|
|
t.Helper()
|
|
for _, name := range []string{"TMPDIR", "TMP", "TEMP"} {
|
|
t.Setenv(name, dir)
|
|
}
|
|
}
|
|
|
|
func assertMirrorLocked(t *testing.T, mirrorRoot string) {
|
|
t.Helper()
|
|
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
|
|
defer cancel()
|
|
lock, err := AcquireMirrorLock(ctx, mirrorRoot)
|
|
if lock != nil {
|
|
require.NoError(t, lock.Close())
|
|
}
|
|
assert.Error(t, err)
|
|
}
|
|
|
|
func assertMirrorUnlocked(t *testing.T, mirrorRoot string) {
|
|
t.Helper()
|
|
lock, err := AcquireMirrorLock(context.Background(), mirrorRoot)
|
|
require.NoError(t, err)
|
|
require.NoError(t, lock.Close())
|
|
}
|