Skip to content

📝 CodeRabbit Chat: Publish archives atomically without replacing existing destinations #158

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
coderabbitai wants to merge 1 commit into from
Closed

📝 CodeRabbit Chat: Publish archives atomically without replacing existing destinations #158

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions cmd/retriever/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,8 +178,9 @@ retriever dump \
The TAR contains the manifest and its JSONL fragments. `-out` remains the
collection directory; `-archive-out` adds a second artifact. Parquet sidecars,
when requested, stay in the directory and are not included in the TAR.
Existing archive files are refused, and partial archive files are removed if
packaging fails.
Existing archive paths, including dangling symlinks, are refused even if
created while packaging is in progress. Publication requires filesystem support
for hard links. Temporary archive files are removed after success or failure.
To load an unencrypted TAR, extract it into a collection directory and pass
that directory to `retriever load -in`. The `retriever unpack` and
`retriever load -archive` commands currently accept encrypted archives only.
Expand Down
6 changes: 3 additions & 3 deletions retriever/archive_envelope.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,10 +159,10 @@ func WriteCollectionArchiveFile(options ArchiveFileOptions) error {
if err := tempFile.Close(); err != nil {
return fmt.Errorf("close archive temp file: %w", err)
}
// Linking publishes the completed file only if the destination is absent.
// Unlike a check followed by Rename, this cannot replace a concurrent output.
// Linking publishes without replacing a destination created after preflight.
// Deferred cleanup removes only the temporary name, leaving the archive intact.
if err := os.Link(tempPath, outputPath); err != nil {
return fmt.Errorf("publish archive: %w", err)
return fmt.Errorf("publish archive %s: %w", outputPath, err)
}

if options.Recipient != nil {
Expand Down
19 changes: 17 additions & 2 deletions retriever/archive_envelope_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,13 +125,25 @@ func TestEncryptedCollectionArchiveSyntheticRoundTrip(t *testing.T) {

dumpDir := writeArchiveFixture(t)
archivePath := filepath.Join(dir, "dump.tar.pq")
completed := false
if err := WriteCollectionArchiveFile(ArchiveFileOptions{
CollectionDir: dumpDir,
OutputPath: archivePath,
Recipient: publicKey,
Progress: func(event ProgressEvent) {
if event.Message == "retriever archive encryption completed" {
completed = true
if _, err := os.ReadFile(archivePath); err != nil {
t.Errorf("archive unavailable at completion: %v", err)
}
}
},
}); err != nil {
t.Fatalf("write encrypted collection archive: %v", err)
}
if !completed {
t.Fatal("expected encryption completion after publication")
}

outputDir := filepath.Join(dir, "unpacked")
if err := unpackEncryptedCollectionArchive(archivePath, outputDir, false, privateKey); err != nil {
Expand All @@ -148,8 +160,11 @@ func TestEncryptedCollectionArchiveSyntheticRoundTrip(t *testing.T) {
t.Fatalf("read unpacked Manifest: %v", err)
}

if _, err := os.Stat(archivePath + ".tmp"); !os.IsNotExist(err) {
t.Fatalf("expected archive temp path cleanup, got %v", err)
if temps, err := filepath.Glob(archivePath + ".*.tmp"); err != nil || len(temps) != 0 {
t.Fatalf("temporary archives remain: %v, error = %v", temps, err)
}
if info, err := os.Stat(archivePath); err != nil || info.Mode().Perm() != 0o600 {
t.Fatalf("archive file mode = %v, error = %v", info, err)
}
}

Expand Down
159 changes: 71 additions & 88 deletions retriever/archive_tar_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,9 @@ func TestCollectionTarFileMatchesStreamAndExcludesSidecars(t *testing.T) {
t.Fatalf("tar contains sidecar %q", name)
}
}
if temps, err := filepath.Glob(archivePath + ".*.tmp"); err != nil || len(temps) != 0 {
t.Fatalf("temporary archives remain: %v, error = %v", temps, err)
}
if info, err := os.Stat(archivePath); err != nil || info.Mode().Perm() != 0o600 {
t.Fatalf("archive file mode = %v, error = %v", info, err)
}
Expand Down Expand Up @@ -130,105 +133,85 @@ func TestCollectionTarFilePreservesExistingOutputAndCleansUpFailure(t *testing.T
}
}

func TestCollectionArchiveFilePublication(t *testing.T) {
privateKey, publicKey := generateTestArchiveKeyPair(t)
func TestCollectionArchiveFilePublicationCollisions(t *testing.T) {
for _, encrypted := range []bool{false, true} {
name := "plain"
if encrypted {
name = "encrypted"
}
t.Run(name, func(t *testing.T) {
for _, test := range []struct {
name string
collision bool
symlink bool
late bool
}{
{name: "success"},
{name: "existing_file", collision: true},
{name: "existing_dangling_symlink", collision: true, symlink: true},
{name: "file_created_during_write", collision: true, late: true},
{name: "dangling_symlink_created_during_write", collision: true, symlink: true, late: true},
} {
t.Run(test.name, func(t *testing.T) {
dumpDir := writeArchiveFixture(t)
archiveDir := t.TempDir()
archivePath := filepath.Join(archiveDir, "dump.tar")
missingTarget := filepath.Join(archiveDir, "missing")
created := false
createCollision := func() {
t.Helper()
var err error
if test.symlink {
err = os.Symlink(missingTarget, archivePath)
} else {
err = os.WriteFile(archivePath, []byte("existing"), 0o600)
}
if err != nil {
t.Fatalf("create collision: %v", err)
for _, afterPreflight := range []bool{false, true} {
phase := "preflight"
if afterPreflight {
phase = "publication"
}
t.Run(phase, func(t *testing.T) {
for _, symlink := range []bool{false, true} {
kind := "file"
if symlink {
kind = "dangling symlink"
}
created = true
}
if test.collision && !test.late {
createCollision()
}
options := ArchiveFileOptions{
CollectionDir: dumpDir,
OutputPath: archivePath,
Progress: func(event ProgressEvent) {
if test.late && event.Message == "retriever archive tar streaming completed" {
createCollision()
t.Run(kind, func(t *testing.T) {
archiveDir := t.TempDir()
archivePath := filepath.Join(archiveDir, "dump.tar")
target := filepath.Join(archiveDir, "missing")
options := ArchiveFileOptions{
CollectionDir: writeArchiveFixture(t),
OutputPath: archivePath,
}
},
}
if encrypted {
options.Recipient = publicKey
}
err := WriteCollectionArchiveFile(options)
if test.collision {
if !created {
t.Fatal("collision was not created")
}
if err == nil {
t.Fatal("expected output collision to fail")
}
if test.late && !errors.Is(err, os.ErrExist) {
t.Fatalf("expected publication to fail with ErrExist, got %v", err)
}
if test.symlink {
if target, err := os.Readlink(archivePath); err != nil || target != missingTarget {
t.Fatalf("symlink changed: target = %q, error = %v", target, err)
if encrypted {
_, options.Recipient = generateTestArchiveKeyPair(t)
}
if _, err := os.Lstat(missingTarget); !os.IsNotExist(err) {
t.Fatalf("symlink target was created: %v", err)
createDestination := func() {
t.Helper()
var err error
if symlink {
err = os.Symlink(target, archivePath)
} else {
err = os.WriteFile(archivePath, []byte("existing"), 0o600)
}
if err != nil {
t.Fatalf("create destination: %v", err)
}
}
} else if contents, err := os.ReadFile(archivePath); err != nil || string(contents) != "existing" {
t.Fatalf("existing output changed: %q, error = %v", contents, err)
}
} else {
if err != nil {
t.Fatalf("write archive: %v", err)
}
contents, err := os.ReadFile(archivePath)
if err != nil {
t.Fatalf("read archive: %v", err)
}
if encrypted {
contents, err = decryptArchiveBytes(contents, privateKey)
if err != nil {
t.Fatalf("decrypt archive: %v", err)
created := false
if afterPreflight {
options.Progress = func(event ProgressEvent) {
if event.Message == "retriever archive tar streaming completed" {
createDestination()
created = true
}
if event.Message == "retriever archive encryption completed" {
t.Error("encryption success reported before publication")
}
}
} else {
createDestination()
}
}
var expected bytes.Buffer
if err := WriteCollectionTar(&expected, dumpDir); err != nil {
t.Fatalf("write expected tar: %v", err)
}
if !bytes.Equal(contents, expected.Bytes()) {
t.Fatal("published archive differs from expected tar")
}
}
if entries, err := os.ReadDir(archiveDir); err != nil || len(entries) != 1 || entries[0].Name() != "dump.tar" {
t.Fatalf("expected only output after cleanup, got %v, error = %v", entries, err)
err := WriteCollectionArchiveFile(options)
if err == nil {
t.Fatal("expected destination collision")
}
if afterPreflight && (!created || !strings.Contains(err.Error(), "publish archive") || !errors.Is(err, os.ErrExist)) {
t.Fatalf("expected atomic publication collision, created = %v, error = %v", created, err)
}
if !afterPreflight && !strings.Contains(err.Error(), "already exists") {
t.Fatalf("expected preflight collision: %v", err)
}
if symlink {
if actual, err := os.Readlink(archivePath); err != nil || actual != target {
t.Fatalf("symlink changed: %q, error = %v", actual, err)
}
if _, err := os.Lstat(target); !os.IsNotExist(err) {
t.Fatalf("symlink target created: %v", err)
}
} else if actual, err := os.ReadFile(archivePath); err != nil || string(actual) != "existing" {
t.Fatalf("destination changed: %q, error = %v", actual, err)
}
if entries, err := os.ReadDir(archiveDir); err != nil || len(entries) != 1 {
t.Fatalf("temporary archives remain: %v, error = %v", entries, err)
}
})
}
})
}
Expand Down