aboutsummaryrefslogtreecommitdiff
path: root/internal/persistence
diff options
context:
space:
mode:
Diffstat (limited to 'internal/persistence')
-rw-r--r--internal/persistence/txn.go60
-rw-r--r--internal/persistence/txn_directory_durability_test.go175
2 files changed, 216 insertions, 19 deletions
diff --git a/internal/persistence/txn.go b/internal/persistence/txn.go
index 9939204..400d355 100644
--- a/internal/persistence/txn.go
+++ b/internal/persistence/txn.go
@@ -64,9 +64,9 @@ type Transaction struct {
ops map[string]*fileOp
order []string
journalDir string
- // dirSync makes a journal directory entry durable (nil means syncDir).
- // It is a field rather than a package-level function so a test can
- // assert that the commit point is synced and inject a sync failure.
+ // dirSync makes directory mutations durable (nil means syncDir).
+ // Per-transaction so tests can check ordering and inject failures
+ // without changing a package-global hook.
dirSync func(string) error
}
@@ -82,12 +82,31 @@ func syncDir(dir string) error {
return d.Sync()
}
-// syncJournalDir syncs the transaction's journal directory.
-func (t *Transaction) syncJournalDir() error {
+func (t *Transaction) syncDirectory(dir string) error {
if t.dirSync != nil {
- return t.dirSync(t.journalDir)
+ return t.dirSync(dir)
}
- return syncDir(t.journalDir)
+ return syncDir(dir)
+}
+
+// removeFileDurable also syncs an already-absent file's parent: recovery may
+// be retrying a deletion whose previous directory sync failed. A missing
+// parent needs no sync because this transaction does not create directories.
+func (t *Transaction) removeFileDurable(path string) error {
+ err := os.Remove(path)
+ if err != nil && !os.IsNotExist(err) {
+ return err
+ }
+ dir := filepath.Dir(path)
+ if os.IsNotExist(err) {
+ if _, statErr := os.Stat(dir); os.IsNotExist(statErr) {
+ return nil
+ }
+ }
+ if err := t.syncDirectory(dir); err != nil {
+ return fmt.Errorf("sync directory %s after delete: %w", dir, err)
+ }
+ return nil
}
// removeJournalDurable unlinks a journal and syncs its directory, so the
@@ -101,7 +120,7 @@ func (t *Transaction) removeJournalDurable(path string) error {
for attempt := 0; attempt < journalRemoveAttempts; attempt++ {
removeErr = os.Remove(path)
if removeErr == nil || os.IsNotExist(removeErr) {
- if err := t.syncJournalDir(); err != nil {
+ if err := t.syncDirectory(t.journalDir); err != nil {
return fmt.Errorf("journal %s removed but its directory could not be synced (a crash may resurrect the journal and roll the transaction back): %w", path, err)
}
return nil
@@ -291,7 +310,7 @@ func (t *Transaction) apply() error {
return fmt.Errorf("write %s: %w", op.Path, err)
}
case kindDelete:
- if err := os.Remove(op.Path); err != nil && !os.IsNotExist(err) {
+ if err := t.removeFileDurable(op.Path); err != nil {
return fmt.Errorf("delete %s: %w", op.Path, err)
}
case kindMove:
@@ -306,6 +325,17 @@ func (t *Transaction) apply() error {
if err := os.Rename(op.Path, op.MoveTo); err != nil {
return fmt.Errorf("move %s: %w", op.Path, err)
}
+ // A cross-directory rename changes both directory entries. Make
+ // both durable before the recovery journal can be retired.
+ srcDir, dstDir := filepath.Dir(op.Path), filepath.Dir(op.MoveTo)
+ if err := t.syncDirectory(srcDir); err != nil {
+ return fmt.Errorf("sync move source directory %s: %w", srcDir, err)
+ }
+ if dstDir != srcDir {
+ if err := t.syncDirectory(dstDir); err != nil {
+ return fmt.Errorf("sync move destination directory %s: %w", dstDir, err)
+ }
+ }
case kindMoveTarget:
// Content replace after the move delivered the source bytes.
if op.NewExists {
@@ -333,21 +363,13 @@ func (t *Transaction) rollback() error {
if op.OldExists {
return WriteFile(path, op.Old, os.FileMode(op.OldMode))
}
- return os.Remove(path)
+ return t.removeFileDurable(path)
}
switch op.Kind {
- case kindMove, kindMoveTarget:
- if err := restore(op.Path); err != nil && !os.IsNotExist(err) {
- record(fmt.Errorf("restore %s: %w", op.Path, err))
- }
- case kindReplace:
+ case kindMove, kindMoveTarget, kindReplace, kindDelete:
if err := restore(op.Path); err != nil {
record(fmt.Errorf("restore %s: %w", op.Path, err))
}
- case kindDelete:
- if err := restore(op.Path); err != nil && !os.IsNotExist(err) {
- record(fmt.Errorf("restore %s: %w", op.Path, err))
- }
}
}
return firstErr
diff --git a/internal/persistence/txn_directory_durability_test.go b/internal/persistence/txn_directory_durability_test.go
new file mode 100644
index 0000000..134580e
--- /dev/null
+++ b/internal/persistence/txn_directory_durability_test.go
@@ -0,0 +1,175 @@
+package persistence
+
+import (
+ "errors"
+ "os"
+ "path/filepath"
+ "reflect"
+ "strings"
+ "testing"
+)
+
+func TestTransactionSyncsChangedDirectoriesBeforeCommit(t *testing.T) {
+ for _, operation := range []string{"delete", "move same directory", "move across directories"} {
+ t.Run(operation, func(t *testing.T) {
+ dir := t.TempDir()
+ journalDir := filepath.Join(dir, ".journal")
+ srcDir := filepath.Join(dir, "source")
+ dstDir := srcDir
+ if operation == "move across directories" {
+ dstDir = filepath.Join(dir, "destination")
+ }
+ src, dst := filepath.Join(srcDir, "a.yaml"), filepath.Join(dstDir, "b.yaml")
+ writeTestFile(t, src, "source")
+ writeTestFile(t, dst, "destination")
+ txn := NewTransaction(journalDir)
+ var err error
+ if operation == "delete" {
+ err = txn.Delete(src)
+ } else {
+ err = txn.Move(src, dst)
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ var synced []string
+ txn.dirSync = func(d string) error {
+ entries, err := ListJournalFiles(journalDir)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if d != journalDir && len(entries) != 1 {
+ t.Fatal("journal retired before data directory sync")
+ }
+ if d == journalDir && len(entries) != 0 {
+ t.Fatal("commit point did not remove journal")
+ }
+ if _, err := os.Stat(src); !os.IsNotExist(err) {
+ t.Fatalf("sync must follow deletion/rename: %v", err)
+ }
+ synced = append(synced, d)
+ return syncDir(d)
+ }
+ if err := txn.Commit(); err != nil {
+ t.Fatal(err)
+ }
+ want := []string{srcDir}
+ if dstDir != srcDir {
+ want = append(want, dstDir)
+ }
+ want = append(want, journalDir)
+ if !reflect.DeepEqual(synced, want) {
+ t.Fatalf("sync order = %v, want %v", synced, want)
+ }
+ if operation != "delete" && readTestFile(t, dst) != "source" {
+ t.Fatal("move did not replace destination")
+ }
+ })
+ }
+}
+
+func TestTransactionDataSyncFailureRollsBack(t *testing.T) {
+ for _, failAt := range []string{"delete", "move source", "move destination"} {
+ t.Run(failAt, func(t *testing.T) {
+ dir := t.TempDir()
+ journalDir := filepath.Join(dir, ".journal")
+ srcDir, dstDir := filepath.Join(dir, "source"), filepath.Join(dir, "destination")
+ src, dst := filepath.Join(srcDir, "a.yaml"), filepath.Join(dstDir, "b.yaml")
+ writeTestFile(t, src, "source")
+ writeTestFile(t, dst, "destination")
+ txn := NewTransaction(journalDir)
+ var err error
+ if failAt == "delete" {
+ err = txn.Delete(src)
+ } else {
+ err = txn.Move(src, dst)
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ failed := false
+ fault := errors.New("injected data directory sync failure")
+ txn.dirSync = func(d string) error {
+ failDir := srcDir
+ if failAt == "move destination" {
+ failDir = dstDir
+ }
+ if d == failDir {
+ failed = true
+ return fault
+ }
+ // Restoring existing before-images uses WriteFile, which
+ // already syncs their directories before journal removal.
+ if d == journalDir && (readTestFile(t, src) != "source" || readTestFile(t, dst) != "destination") {
+ t.Fatal("journal retired before rollback restored files")
+ }
+ return syncDir(d)
+ }
+ if err := txn.Commit(); !errors.Is(err, fault) || !failed {
+ t.Fatalf("commit error = %v, fault injected = %t", err, failed)
+ }
+ entries, err := ListJournalFiles(journalDir)
+ if err != nil || len(entries) != 0 {
+ t.Fatalf("successful rollback should retire journal: %v, %v", entries, err)
+ }
+ })
+ }
+}
+
+func TestRollbackDeletionSyncFailureRetainsJournalAndRetries(t *testing.T) {
+ dir := t.TempDir()
+ journalDir := filepath.Join(dir, ".journal")
+ createdDir, failDir := filepath.Join(dir, "created"), filepath.Join(dir, "fail")
+ if err := os.Mkdir(createdDir, 0700); err != nil {
+ t.Fatal(err)
+ }
+ created, deleted := filepath.Join(createdDir, "new.yaml"), filepath.Join(failDir, "old.yaml")
+ writeTestFile(t, deleted, "before")
+ txn := NewTransaction(journalDir)
+ if err := txn.Replace(created, []byte("new"), 0600); err != nil {
+ t.Fatal(err)
+ }
+ if err := txn.Delete(deleted); err != nil {
+ t.Fatal(err)
+ }
+ var synced []string
+ txn.dirSync = func(d string) error {
+ synced = append(synced, d)
+ return errors.New("injected directory sync failure")
+ }
+ if err := txn.Commit(); err == nil || !strings.Contains(err.Error(), "rollback failed") {
+ t.Fatalf("expected apply and rollback sync failures, got %v", err)
+ }
+ if !reflect.DeepEqual(synced, []string{failDir, createdDir}) {
+ t.Fatalf("syncs = %v; missing rollback deletion barrier", synced)
+ }
+ if _, err := os.Stat(created); !os.IsNotExist(err) {
+ t.Fatalf("rollback should have removed created file: %v", err)
+ }
+ if entries, err := ListJournalFiles(journalDir); err != nil || len(entries) != 1 {
+ t.Fatalf("failed rollback must retain journal: %v, %v", entries, err)
+ }
+
+ // Retry the same rollback with the file already absent: skipping its
+ // directory sync would incorrectly retire an undurable deletion.
+ synced = nil
+ txn.dirSync = func(d string) error {
+ synced = append(synced, d)
+ return syncDir(d)
+ }
+ if err := txn.rollback(); err != nil {
+ t.Fatal(err)
+ }
+ if !reflect.DeepEqual(synced, []string{createdDir}) {
+ t.Fatalf("retry syncs = %v, want %s", synced, createdDir)
+ }
+ if err := Recover(journalDir); err != nil {
+ t.Fatal(err)
+ }
+ if readTestFile(t, deleted) != "before" {
+ t.Fatal("recovery did not restore deleted file")
+ }
+ if entries, err := ListJournalFiles(journalDir); err != nil || len(entries) != 0 {
+ t.Fatalf("recovery did not retire journal: %v, %v", entries, err)
+ }
+}
generated by cgit v1.2.3 (git 2.25.1) at 2026-10-24 12:13:29 +0000