Sync agent queue replacements before success

Agent queue writes sync the replacement file and its directory before
returning success. This gives local event acceptance a durable storage
boundary and stops queue writes when a replacement has an uncertain
outcome.

Issue: 44d8703

diff --git a/daemon_message.go b/daemon_message.go
index 7665678468186a29b88e53ef38b1ddbad79d57e0..59733be7022e89ce73fe433afe2866ff16651c4a 100644 (file)
--- a/daemon_message.go
+++ b/daemon_message.go
@@ -193,7 +193,7 @@ func (m *DaemonMessages) save() error {
                return fmt.Errorf("encode agent messages: %w", err)
        }
 
                return fmt.Errorf("encode agent messages: %w", err)
        }
 

-       err = writeFileAtomic(m.path, append(data, '\n'))
+       err = writeFileDurable(m.path, append(data, '\n'))

        if err != nil {
                return fmt.Errorf("save agent messages: %w", err)
        }
        if err != nil {
                return fmt.Errorf("save agent messages: %w", err)
        }

diff --git a/file.go b/file.go
index a7ba5bbdd78b6ec54d1ff6ab599dec2eb78212e7..00c8a0cb9509881e5a4f632d8c9f16c943269ea5 100644 (file)
--- a/file.go
+++ b/file.go
@@ -28,3 +28,47 @@ func writeFileAtomic(path string, data []byte) error {
 
        return nil
 }
 
        return nil
 }

+
+// writeFileDurable replaces path only after syncing its private temporary file.
+// It syncs the directory after rename before it reports success. An error after
+// rename has an uncertain outcome, so callers must stop writes until recovery.
+func writeFileDurable(path string, data []byte) (err error) {
+       file, err := os.CreateTemp(filepath.Dir(path), "."+filepath.Base(path)+".*")
+       if err != nil {
+               return fmt.Errorf("create durable temporary file: %w", err)
+       }
+
+       defer func() {
+               removeErr := os.Remove(file.Name())
+               if removeErr != nil && !errors.Is(removeErr, os.ErrNotExist) {
+                       err = errors.Join(err, fmt.Errorf("remove durable temporary file: %w", removeErr))
+               }
+       }()
+
+       _, err = file.Write(data)
+       if err == nil {
+               err = file.Sync()
+       }
+
+       err = errors.Join(err, file.Close())
+       if err != nil {
+               return fmt.Errorf("write durable temporary file: %w", err)
+       }
+
+       err = os.Rename(file.Name(), path)
+       if err != nil {
+               return fmt.Errorf("publish durable file: %w", err)
+       }
+
+       directory, err := os.Open(filepath.Dir(path))
+       if err != nil {
+               return fmt.Errorf("open durable directory: %w", err)
+       }
+
+       err = errors.Join(directory.Sync(), directory.Close())
+       if err != nil {
+               return fmt.Errorf("sync durable directory: %w", err)
+       }
+
+       return nil
+}

diff --git a/file_test.go b/file_test.go
new file mode 100644 (file)
index 0000000..e07d484
--- /dev/null
+++ b/file_test.go
@@ -0,0 +1,40 @@
+package main
+
+import (
+       "os"
+       "path/filepath"
+       "testing"
+
+       "github.com/stretchr/testify/require"
+)
+
+func TestDurableReplacementKeepsPrivateCompleteContents(t *testing.T) {
+       path := filepath.Join(t.TempDir(), "state.json")
+       require.NoError(t, writeFileDurable(path, []byte("first\n")))
+       require.NoError(t, writeFileDurable(path, []byte("second\n")))
+
+       data, err := os.ReadFile(path)
+       require.NoError(t, err)
+       require.Equal(t, "second\n", string(data))
+
+       info, err := os.Stat(path)
+       require.NoError(t, err)
+       require.Equal(t, os.FileMode(0o600), info.Mode().Perm())
+
+       entries, err := os.ReadDir(filepath.Dir(path))
+       require.NoError(t, err)
+       require.Len(t, entries, 1)
+}
+
+func TestDurableReplacementReportsFailuresAndRemovesTemporaryFiles(t *testing.T) {
+       directory := t.TempDir()
+       path := filepath.Join(directory, "state.json")
+       require.NoError(t, os.Mkdir(path, 0o700))
+       require.ErrorContains(t, writeFileDurable(path, []byte("data")), "publish durable file")
+
+       entries, err := os.ReadDir(directory)
+       require.NoError(t, err)
+       require.Len(t, entries, 1)
+       require.True(t, entries[0].IsDir())
+       require.ErrorContains(t, writeFileDurable(filepath.Join(directory, "missing", "state"), nil), "create durable temporary file")
+}