From c1caed3d42b25cd558a037a814c879fe7483f0c1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Thu, 3 Sep 2026 01:20:38 -0400 Subject: [PATCH] chore(go): audit transaction engine Share change filtering between the durable and test transaction paths, collapse repeated precondition handling, unify journal sequence parsing, and split the subprocess worker test file. Claude-Session: https://claude.ai/code/session_01Agoc9bRVE9KL7mmWtey9Vq --- cmd/elydora/plugins/transaction.go | 90 ++---- cmd/elydora/plugins/transaction_journal.go | 23 +- .../plugins/transaction_subprocess_test.go | 276 ----------------- .../transaction_subprocess_worker_test.go | 287 ++++++++++++++++++ 4 files changed, 328 insertions(+), 348 deletions(-) create mode 100644 cmd/elydora/plugins/transaction_subprocess_worker_test.go diff --git a/cmd/elydora/plugins/transaction.go b/cmd/elydora/plugins/transaction.go index 619f870..a270a8e 100644 --- a/cmd/elydora/plugins/transaction.go +++ b/cmd/elydora/plugins/transaction.go @@ -5,9 +5,7 @@ import ( "errors" "fmt" "os" - "path/filepath" "runtime" - "strings" ) type renameFunc func(source, destination string) error @@ -165,84 +163,60 @@ func writeChangesWithFileOps( if ops.moveNoReplace == nil { return fmt.Errorf("%s is missing atomic file operations", label) } - filtered := make([]fileChange, 0, len(changes)) - targets := map[string]bool{} - for _, change := range changes { - if change == nil || (change.existed && !change.remove && - bytes.Equal(change.original, change.next) && - sameManagedFileMode(change.originalMode, change.mode)) { - continue - } - target := filepath.Clean(change.filePath) - if runtime.GOOS == "windows" { - target = strings.ToLower(target) - } - if targets[target] { - return fmt.Errorf("%s contains duplicate file target %s", label, change.filePath) - } - targets[target] = true - filtered = append(filtered, *change) + filtered, err := filterDurableChanges(changes, label) + if err != nil { + return err } - if err := assertFilePreconditions(preconditions, label); err != nil { - return fmt.Errorf("%s: %w", label, err) + checkPreconditions := func() error { + if err := assertFilePreconditions(preconditions, label); err != nil { + return fmt.Errorf("%s: %w", label, err) + } + return nil + } + if err := checkPreconditions(); err != nil { + return err } if len(filtered) == 0 { return nil } staged := make([]stagedChange, 0, len(filtered)) + abortStaging := func(cause error) error { + return joinTransactionFailure(cause, cleanupStaging(staged), "cleanup failed") + } for _, change := range filtered { - if err := assertFilePreconditions(preconditions, label); err != nil { - cleanupErrors := cleanupStaging(staged) - return joinTransactionFailure( - fmt.Errorf("%s: %w", label, err), - cleanupErrors, - "cleanup failed", - ) + if err := checkPreconditions(); err != nil { + return abortStaging(err) } item, err := stageChange(change) if err != nil { - cleanupErrors := cleanupStaging(staged) - return joinTransactionFailure(fmt.Errorf("%s: %w", label, err), cleanupErrors, "cleanup failed") + return abortStaging(fmt.Errorf("%s: %w", label, err)) } staged = append(staged, item) - if err := assertFilePreconditions(preconditions, label); err != nil { - cleanupErrors := cleanupStaging(staged) - return joinTransactionFailure( - fmt.Errorf("%s: %w", label, err), - cleanupErrors, - "cleanup failed", - ) - } } - if err := assertFilePreconditions(preconditions, label); err != nil { - cleanupErrors := cleanupStaging(staged) - return joinTransactionFailure(fmt.Errorf("%s: %w", label, err), cleanupErrors, "cleanup failed") + if err := checkPreconditions(); err != nil { + return abortStaging(err) + } + recoverStaged := func(cause error) error { + failures := rollbackChanges(staged, ops) + failures = append(failures, cleanupStaging(staged)...) + return joinTransactionFailure(cause, failures, "recovery failed") } for index := range staged { - if err := assertFilePreconditions(preconditions, label); err != nil { - recoveryErrors := rollbackChanges(staged, ops) - recoveryErrors = append(recoveryErrors, cleanupStaging(staged)...) - return joinTransactionFailure(fmt.Errorf("%s: %w", label, err), recoveryErrors, "recovery failed") + if err := checkPreconditions(); err != nil { + return recoverStaged(err) } if err := commitChange(&staged[index], ops); err != nil { - recoveryErrors := rollbackChanges(staged, ops) - recoveryErrors = append(recoveryErrors, cleanupStaging(staged)...) - return joinTransactionFailure(fmt.Errorf("%s: %w", label, err), recoveryErrors, "recovery failed") + return recoverStaged(fmt.Errorf("%s: %w", label, err)) } } - if err := assertFilePreconditions(preconditions, label); err != nil { - recoveryErrors := rollbackChanges(staged, ops) - recoveryErrors = append(recoveryErrors, cleanupStaging(staged)...) - return joinTransactionFailure(fmt.Errorf("%s: %w", label, err), recoveryErrors, "recovery failed") + if err := checkPreconditions(); err != nil { + return recoverStaged(err) } if err := assertCommittedChanges(staged); err != nil { - recoveryErrors := rollbackChanges(staged, ops) - recoveryErrors = append(recoveryErrors, cleanupStaging(staged)...) - return joinTransactionFailure(fmt.Errorf("%s: %w", label, err), recoveryErrors, "recovery failed") + return recoverStaged(fmt.Errorf("%s: %w", label, err)) } - cleanupErrors := cleanupStaging(staged) - if len(cleanupErrors) > 0 { - return joinTransactionFailure(fmt.Errorf("%s cleanup failed", label), cleanupErrors, "cleanup failed") + if failures := cleanupStaging(staged); len(failures) > 0 { + return joinTransactionFailure(fmt.Errorf("%s cleanup failed", label), failures, "cleanup failed") } return nil } diff --git a/cmd/elydora/plugins/transaction_journal.go b/cmd/elydora/plugins/transaction_journal.go index 65042f6..1c86388 100644 --- a/cmd/elydora/plugins/transaction_journal.go +++ b/cmd/elydora/plugins/transaction_journal.go @@ -178,8 +178,7 @@ func loadDurableJournal(directory string) (*durableJournal, error) { } stateNames := make([]string, 0) for _, entry := range entries { - if !entry.IsDir() && strings.HasPrefix(entry.Name(), "state-") && - strings.HasSuffix(entry.Name(), ".json") { + if _, state := durableJournalStateSequence(entry.Name()); state && !entry.IsDir() { stateNames = append(stateNames, entry.Name()) } } @@ -317,11 +316,11 @@ func validateDurableJournalProgress(journal *durableJournal) error { return nil } -func durableJournalStateSequence(name string) (uint64, bool) { - if !strings.HasPrefix(name, "state-") || !strings.HasSuffix(name, ".json") { +func durableJournalSequence(name, suffix string) (uint64, bool) { + if !strings.HasPrefix(name, "state-") || !strings.HasSuffix(name, suffix) { return 0, false } - value := strings.TrimSuffix(strings.TrimPrefix(name, "state-"), ".json") + value := strings.TrimSuffix(strings.TrimPrefix(name, "state-"), suffix) if len(value) != 20 { return 0, false } @@ -329,14 +328,10 @@ func durableJournalStateSequence(name string) (uint64, bool) { return sequence, err == nil } +func durableJournalStateSequence(name string) (uint64, bool) { + return durableJournalSequence(name, ".json") +} + func durableJournalPendingSequence(name string) (uint64, bool) { - if !strings.HasPrefix(name, "state-") || !strings.HasSuffix(name, ".pending") { - return 0, false - } - value := strings.TrimSuffix(strings.TrimPrefix(name, "state-"), ".pending") - if len(value) != 20 { - return 0, false - } - sequence, err := strconv.ParseUint(value, 10, 64) - return sequence, err == nil + return durableJournalSequence(name, ".pending") } diff --git a/cmd/elydora/plugins/transaction_subprocess_test.go b/cmd/elydora/plugins/transaction_subprocess_test.go index 302f7e9..4775580 100644 --- a/cmd/elydora/plugins/transaction_subprocess_test.go +++ b/cmd/elydora/plugins/transaction_subprocess_test.go @@ -13,282 +13,6 @@ import ( "time" ) -const durableWorkerEnvironment = "ELYDORA_DURABLE_TRANSACTION_WORKER" - -func TestDurableTransactionSubprocessWorker(t *testing.T) { - mode := os.Getenv(durableWorkerEnvironment) - if mode == "" { - return - } - target := os.Getenv("ELYDORA_DURABLE_TARGET") - switch mode { - case "crash-update": - runDurableCrashWorker(t, target) - case "crash-multi-update": - runDurableMultiCrashWorker(t, target, os.Getenv("ELYDORA_DURABLE_SECOND_TARGET")) - case "crash-create": - runDurableCreateCrashWorker(t, target) - case "crash-delete": - runDurableDeleteCrashWorker(t, target) - case "crash-committed-cleanup": - runDurableCommittedCleanupCrashWorker(t, target) - case "crash-rollback-hardlink": - runDurableRollbackHardlinkCrashWorker(t) - case "crash-init-workspace-empty": - runDurableInitializationWorkspaceCrashWorker(t, target, false) - case "crash-init-workspace-owner": - runDurableInitializationWorkspaceCrashWorker(t, target, true) - case "crash-init-stage-full": - runDurableInitializationStageCrashWorker(t, target, false) - case "crash-init-stage-partial": - runDurableInitializationStageCrashWorker(t, target, true) - case "crash-init-probe": - runDurableInitializationProbeCrashWorker(t, target) - case "crash-journal-owner": - runDurableJournalOwnerCrashWorker(t) - case "crash-journal-pending-full": - runDurableJournalPendingCrashWorker(t, target, false) - case "crash-journal-pending-partial": - runDurableJournalPendingCrashWorker(t, target, true) - case "crash-workspace-marker-cleanup": - runDurableWorkspaceMarkerCleanupCrashWorker(t, target) - case "crash-journal-marker-cleanup": - runDurableJournalMarkerCleanupCrashWorker(t, target) - case "crash-torn-journal-tail": - runDurableTornJournalTailCrashWorker(t, target) - case "reader": - runDurableReaderWorker(t, target) - case "writer": - runDurableWriterWorker(t, target) - default: - t.Fatalf("unknown durable transaction worker %q", mode) - } -} - -func runDurableRollbackHardlinkCrashWorker(t *testing.T) { - stateRoot, err := durableTransactionStateRoot() - if err != nil { - t.Fatal(err) - } - entries, err := os.ReadDir(stateRoot) - if err != nil { - t.Fatal(err) - } - journalPath := "" - for _, entry := range entries { - if entry.IsDir() && strings.HasPrefix(entry.Name(), "txn-") { - if journalPath != "" { - t.Fatal("multiple recovery journals found") - } - journalPath = filepath.Join(stateRoot, entry.Name()) - } - } - if journalPath == "" { - t.Fatal("recovery journal is unavailable") - } - journal, err := loadDurableJournal(journalPath) - if err != nil || len(journal.Entries) != 1 { - t.Fatalf("load recovery journal: %v", err) - } - entry := journal.Entries[0] - if err := os.Link(entry.Path, entry.DiscardPath); err != nil { - t.Fatal(err) - } - if err := syncTransactionDirectory(filepath.Dir(entry.DiscardPath)); err != nil { - t.Fatal(err) - } - os.Exit(91) -} - -func runDurableMultiCrashWorker(t *testing.T, firstTarget, secondTarget string) { - changes := make([]*fileChange, 0, 2) - for _, item := range []struct { - path string - value string - }{{firstTarget, "first-next"}, {secondTarget, "second-next"}} { - change, err := prepareFileChange(item.path, "multi crash target", []byte(item.value), 0600) - if err != nil { - t.Fatal(err) - } - changes = append(changes, change) - } - ops := newTransactionFileOps(nil) - replace := ops.replaceWithBackup - commits := 0 - ops.replaceWithBackup = func(current, replacement, backup string) error { - if err := replace(current, replacement, backup); err != nil { - return err - } - commits++ - if commits == 2 { - os.Exit(87) - } - return nil - } - if err := writeChangesWithFileOps(changes, "multi crash transaction", ops); err != nil { - t.Fatal(err) - } - t.Fatal("multi crash worker completed without terminating") -} - -func runDurableCreateCrashWorker(t *testing.T, target string) { - change, err := prepareFileChange(target, "crash create target", []byte("created"), 0600) - if err != nil { - t.Fatal(err) - } - ops := newTransactionFileOps(nil) - move := ops.moveNoReplace - ops.moveNoReplace = func(source, destination string) error { - if err := move(source, destination); err != nil { - return err - } - if sameTransactionTestPath(destination, target) { - os.Exit(88) - } - return nil - } - if err := writeChangesWithFileOps([]*fileChange{change}, "crash create", ops); err != nil { - t.Fatal(err) - } - t.Fatal("create crash worker completed without terminating") -} - -func runDurableDeleteCrashWorker(t *testing.T, target string) { - change, err := prepareSourceChange( - target, - "crash delete target", - []byte("delete-original"), - true, - nil, - 0600, - true, - ) - if err != nil { - t.Fatal(err) - } - ops := newTransactionFileOps(nil) - move := ops.moveNoReplace - ops.moveNoReplace = func(source, destination string) error { - if err := move(source, destination); err != nil { - return err - } - if sameTransactionTestPath(source, target) { - os.Exit(89) - } - return nil - } - if err := writeChangesWithFileOps([]*fileChange{change}, "crash delete", ops); err != nil { - t.Fatal(err) - } - t.Fatal("delete crash worker completed without terminating") -} - -func runDurableCommittedCleanupCrashWorker(t *testing.T, target string) { - change, err := prepareFileChange(target, "committed cleanup target", []byte("committed"), 0600) - if err != nil { - t.Fatal(err) - } - ops := newTransactionFileOps(nil) - replace := ops.replaceWithBackup - blocker := "" - ops.replaceWithBackup = func(current, replacement, backup string) error { - if err := replace(current, replacement, backup); err != nil { - return err - } - blocker = filepath.Join(filepath.Dir(backup), "cleanup-blocker") - return os.WriteFile(blocker, []byte("block"), 0600) - } - err = writeChangesWithFileOps([]*fileChange{change}, "committed cleanup", ops) - if err == nil || !strings.Contains(err.Error(), "committed; durable cleanup failed") { - t.Fatalf("committed cleanup error = %v", err) - } - if err := os.Remove(blocker); err != nil { - t.Fatal(err) - } - os.Exit(90) -} - -func runDurableCrashWorker(t *testing.T, target string) { - change, err := prepareFileChange(target, "crash target", []byte("next"), 0600) - if err != nil { - t.Fatal(err) - } - ops := newTransactionFileOps(nil) - replace := ops.replaceWithBackup - absoluteTarget, err := filepath.Abs(target) - if err != nil { - t.Fatal(err) - } - ops.replaceWithBackup = func(current, replacement, backup string) error { - if err := replace(current, replacement, backup); err != nil { - return err - } - if sameTransactionTestPath(current, absoluteTarget) { - os.Exit(86) - } - return nil - } - if err := writeChangesWithFileOps([]*fileChange{change}, "crash transaction", ops); err != nil { - t.Fatal(err) - } - t.Fatal("crash worker completed without terminating") -} - -func runDurableReaderWorker(t *testing.T, target string) { - stop := os.Getenv("ELYDORA_DURABLE_STOP") - for { - if _, err := os.Lstat(stop); err == nil { - return - } else if !errors.Is(err, os.ErrNotExist) { - t.Fatal(err) - } - content, err := os.ReadFile(target) - if err != nil { - if errors.Is(err, os.ErrNotExist) { - t.Fatalf("reader observed missing target: %v", err) - } - if _, statErr := os.Lstat(target); statErr != nil { - t.Fatalf("reader observed unavailable target: %v; stat=%v", err, statErr) - } - time.Sleep(time.Millisecond) - continue - } - if string(content) != "value-a" && string(content) != "value-b" { - t.Fatalf("reader observed partial value %q", content) - } - time.Sleep(time.Millisecond) - } -} - -func runDurableWriterWorker(t *testing.T, target string) { - value := os.Getenv("ELYDORA_DURABLE_VALUE") - ready := os.Getenv("ELYDORA_DURABLE_READY") - start := os.Getenv("ELYDORA_DURABLE_START") - change, err := prepareFileChange(target, "writer target", []byte(value), 0600) - if err != nil { - t.Fatal(err) - } - if err := os.WriteFile(ready, []byte("ready"), 0600); err != nil { - t.Fatal(err) - } - deadline := time.Now().Add(15 * time.Second) - for { - if _, err := os.Lstat(start); err == nil { - break - } else if !errors.Is(err, os.ErrNotExist) { - t.Fatal(err) - } - if time.Now().After(deadline) { - t.Fatal("writer start barrier timed out") - } - time.Sleep(time.Millisecond) - } - if err := writeChanges([]*fileChange{change}, "subprocess writer", nil); err != nil { - fmt.Fprintln(os.Stderr, err) - os.Exit(23) - } -} - func TestDurableTransactionRecoversAfterProcessTermination(t *testing.T) { home := filepath.Join(t.TempDir(), "home") if err := createPrivateTransactionDirectory(home); err != nil { diff --git a/cmd/elydora/plugins/transaction_subprocess_worker_test.go b/cmd/elydora/plugins/transaction_subprocess_worker_test.go new file mode 100644 index 0000000..fd3b7d1 --- /dev/null +++ b/cmd/elydora/plugins/transaction_subprocess_worker_test.go @@ -0,0 +1,287 @@ +package plugins + +import ( + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +const durableWorkerEnvironment = "ELYDORA_DURABLE_TRANSACTION_WORKER" + +func TestDurableTransactionSubprocessWorker(t *testing.T) { + mode := os.Getenv(durableWorkerEnvironment) + if mode == "" { + return + } + target := os.Getenv("ELYDORA_DURABLE_TARGET") + switch mode { + case "crash-update": + runDurableCrashWorker(t, target) + case "crash-multi-update": + runDurableMultiCrashWorker(t, target, os.Getenv("ELYDORA_DURABLE_SECOND_TARGET")) + case "crash-create": + runDurableCreateCrashWorker(t, target) + case "crash-delete": + runDurableDeleteCrashWorker(t, target) + case "crash-committed-cleanup": + runDurableCommittedCleanupCrashWorker(t, target) + case "crash-rollback-hardlink": + runDurableRollbackHardlinkCrashWorker(t) + case "crash-init-workspace-empty": + runDurableInitializationWorkspaceCrashWorker(t, target, false) + case "crash-init-workspace-owner": + runDurableInitializationWorkspaceCrashWorker(t, target, true) + case "crash-init-stage-full": + runDurableInitializationStageCrashWorker(t, target, false) + case "crash-init-stage-partial": + runDurableInitializationStageCrashWorker(t, target, true) + case "crash-init-probe": + runDurableInitializationProbeCrashWorker(t, target) + case "crash-journal-owner": + runDurableJournalOwnerCrashWorker(t) + case "crash-journal-pending-full": + runDurableJournalPendingCrashWorker(t, target, false) + case "crash-journal-pending-partial": + runDurableJournalPendingCrashWorker(t, target, true) + case "crash-workspace-marker-cleanup": + runDurableWorkspaceMarkerCleanupCrashWorker(t, target) + case "crash-journal-marker-cleanup": + runDurableJournalMarkerCleanupCrashWorker(t, target) + case "crash-torn-journal-tail": + runDurableTornJournalTailCrashWorker(t, target) + case "reader": + runDurableReaderWorker(t, target) + case "writer": + runDurableWriterWorker(t, target) + default: + t.Fatalf("unknown durable transaction worker %q", mode) + } +} + +func runDurableRollbackHardlinkCrashWorker(t *testing.T) { + stateRoot, err := durableTransactionStateRoot() + if err != nil { + t.Fatal(err) + } + entries, err := os.ReadDir(stateRoot) + if err != nil { + t.Fatal(err) + } + journalPath := "" + for _, entry := range entries { + if entry.IsDir() && strings.HasPrefix(entry.Name(), "txn-") { + if journalPath != "" { + t.Fatal("multiple recovery journals found") + } + journalPath = filepath.Join(stateRoot, entry.Name()) + } + } + if journalPath == "" { + t.Fatal("recovery journal is unavailable") + } + journal, err := loadDurableJournal(journalPath) + if err != nil || len(journal.Entries) != 1 { + t.Fatalf("load recovery journal: %v", err) + } + entry := journal.Entries[0] + if err := os.Link(entry.Path, entry.DiscardPath); err != nil { + t.Fatal(err) + } + if err := syncTransactionDirectory(filepath.Dir(entry.DiscardPath)); err != nil { + t.Fatal(err) + } + os.Exit(91) +} + +func runDurableMultiCrashWorker(t *testing.T, firstTarget, secondTarget string) { + changes := make([]*fileChange, 0, 2) + for _, item := range []struct { + path string + value string + }{{firstTarget, "first-next"}, {secondTarget, "second-next"}} { + change, err := prepareFileChange(item.path, "multi crash target", []byte(item.value), 0600) + if err != nil { + t.Fatal(err) + } + changes = append(changes, change) + } + ops := newTransactionFileOps(nil) + replace := ops.replaceWithBackup + commits := 0 + ops.replaceWithBackup = func(current, replacement, backup string) error { + if err := replace(current, replacement, backup); err != nil { + return err + } + commits++ + if commits == 2 { + os.Exit(87) + } + return nil + } + if err := writeChangesWithFileOps(changes, "multi crash transaction", ops); err != nil { + t.Fatal(err) + } + t.Fatal("multi crash worker completed without terminating") +} + +func runDurableCreateCrashWorker(t *testing.T, target string) { + change, err := prepareFileChange(target, "crash create target", []byte("created"), 0600) + if err != nil { + t.Fatal(err) + } + ops := newTransactionFileOps(nil) + move := ops.moveNoReplace + ops.moveNoReplace = func(source, destination string) error { + if err := move(source, destination); err != nil { + return err + } + if sameTransactionTestPath(destination, target) { + os.Exit(88) + } + return nil + } + if err := writeChangesWithFileOps([]*fileChange{change}, "crash create", ops); err != nil { + t.Fatal(err) + } + t.Fatal("create crash worker completed without terminating") +} + +func runDurableDeleteCrashWorker(t *testing.T, target string) { + change, err := prepareSourceChange( + target, + "crash delete target", + []byte("delete-original"), + true, + nil, + 0600, + true, + ) + if err != nil { + t.Fatal(err) + } + ops := newTransactionFileOps(nil) + move := ops.moveNoReplace + ops.moveNoReplace = func(source, destination string) error { + if err := move(source, destination); err != nil { + return err + } + if sameTransactionTestPath(source, target) { + os.Exit(89) + } + return nil + } + if err := writeChangesWithFileOps([]*fileChange{change}, "crash delete", ops); err != nil { + t.Fatal(err) + } + t.Fatal("delete crash worker completed without terminating") +} + +func runDurableCommittedCleanupCrashWorker(t *testing.T, target string) { + change, err := prepareFileChange(target, "committed cleanup target", []byte("committed"), 0600) + if err != nil { + t.Fatal(err) + } + ops := newTransactionFileOps(nil) + replace := ops.replaceWithBackup + blocker := "" + ops.replaceWithBackup = func(current, replacement, backup string) error { + if err := replace(current, replacement, backup); err != nil { + return err + } + blocker = filepath.Join(filepath.Dir(backup), "cleanup-blocker") + return os.WriteFile(blocker, []byte("block"), 0600) + } + err = writeChangesWithFileOps([]*fileChange{change}, "committed cleanup", ops) + if err == nil || !strings.Contains(err.Error(), "committed; durable cleanup failed") { + t.Fatalf("committed cleanup error = %v", err) + } + if err := os.Remove(blocker); err != nil { + t.Fatal(err) + } + os.Exit(90) +} + +func runDurableCrashWorker(t *testing.T, target string) { + change, err := prepareFileChange(target, "crash target", []byte("next"), 0600) + if err != nil { + t.Fatal(err) + } + ops := newTransactionFileOps(nil) + replace := ops.replaceWithBackup + absoluteTarget, err := filepath.Abs(target) + if err != nil { + t.Fatal(err) + } + ops.replaceWithBackup = func(current, replacement, backup string) error { + if err := replace(current, replacement, backup); err != nil { + return err + } + if sameTransactionTestPath(current, absoluteTarget) { + os.Exit(86) + } + return nil + } + if err := writeChangesWithFileOps([]*fileChange{change}, "crash transaction", ops); err != nil { + t.Fatal(err) + } + t.Fatal("crash worker completed without terminating") +} + +func runDurableReaderWorker(t *testing.T, target string) { + stop := os.Getenv("ELYDORA_DURABLE_STOP") + for { + if _, err := os.Lstat(stop); err == nil { + return + } else if !errors.Is(err, os.ErrNotExist) { + t.Fatal(err) + } + content, err := os.ReadFile(target) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + t.Fatalf("reader observed missing target: %v", err) + } + if _, statErr := os.Lstat(target); statErr != nil { + t.Fatalf("reader observed unavailable target: %v; stat=%v", err, statErr) + } + time.Sleep(time.Millisecond) + continue + } + if string(content) != "value-a" && string(content) != "value-b" { + t.Fatalf("reader observed partial value %q", content) + } + time.Sleep(time.Millisecond) + } +} + +func runDurableWriterWorker(t *testing.T, target string) { + value := os.Getenv("ELYDORA_DURABLE_VALUE") + ready := os.Getenv("ELYDORA_DURABLE_READY") + start := os.Getenv("ELYDORA_DURABLE_START") + change, err := prepareFileChange(target, "writer target", []byte(value), 0600) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(ready, []byte("ready"), 0600); err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(15 * time.Second) + for { + if _, err := os.Lstat(start); err == nil { + break + } else if !errors.Is(err, os.ErrNotExist) { + t.Fatal(err) + } + if time.Now().After(deadline) { + t.Fatal("writer start barrier timed out") + } + time.Sleep(time.Millisecond) + } + if err := writeChanges([]*fileChange{change}, "subprocess writer", nil); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(23) + } +}