make statestore creation of meta.json atomic (#51897)

* make statestore creation of meta.json atomic

* add changelog

* fix linter issues

* address scenario that Anderson caught

* fix linter issue

(cherry picked from commit 052272d)

Author: leehinman

changelog/fragments/1783957443-statestore-atomic-create-meta.json.yaml

@@ -0,0 +1,7 @@
+kind: bug-fix
+
+summary: Prevent statestore startup failure when meta.json is left empty after an unclean shutdown
+
+component: libbeat
+
+pr: https://github.com/elastic/beats/pull/51897

libbeat/statestore/backend/memlog/diskstore.go

@@ -29,6 +29,7 @@ import (
 	"strconv"
 
 	"github.com/elastic/beats/v7/libbeat/common/cleanup"
+	agentfile "github.com/elastic/elastic-agent-libs/file"
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/elastic-agent-libs/mapstr"
 )
@@ -195,7 +196,7 @@ func (s *diskstore) tryOpenLog() error {
 			return err
 		}
 
-		s.logFileSize = uint64(info.Size())
+		s.logFileSize = uint64(info.Size()) //nolint:gosec // G115: info.Size() is non-negative for a regular file
 	}
 
 	ok = true
@@ -640,31 +641,35 @@ func checkMeta(meta storeMeta) error {
 
 func writeMetaFile(home string, mode os.FileMode) error {
 	path := filepath.Join(home, metaFileName)
-	f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR|os.O_TRUNC, mode)
+	f, err := os.CreateTemp(home, metaFileName+".tmp-*")
 	if err != nil {
-		return err
+		return fmt.Errorf("error creating tmp meta file: %w", err)
 	}
+	tmpName := f.Name()
+	defer func() {
+		_ = f.Close()
+		_ = os.Remove(tmpName)
+	}()
 
-	ok := false
-	defer cleanup.IfNot(&ok, func() {
-		f.Close()
-	})
+	if err = f.Chmod(mode); err != nil {
+		return fmt.Errorf("error setting tmp meta file permissions: %w", err)
+	}
 
 	enc := newJSONEncoder(f)
-	err = enc.Encode(storeMeta{
-		Version: storeVersion,
-	})
-	if err != nil {
-		return err
+	if err = enc.Encode(storeMeta{Version: storeVersion}); err != nil {
+		return fmt.Errorf("error encoding meta file: %w", err)
 	}
 
-	if err := f.Sync(); err != nil {
-		return err
+	if err = f.Sync(); err != nil {
+		return fmt.Errorf("error syncing meta file: %w", err)
 	}
 
-	ok = true
-	if err := f.Close(); err != nil {
-		return err
+	if err = f.Close(); err != nil {
+		return fmt.Errorf("error closing tmp meta file: %w", err)
+	}
+
+	if err = agentfile.SafeFileRotate(path, tmpName); err != nil {
+		return fmt.Errorf("error rotating meta file into place: %w", err)
 	}
 
 	trySyncPath(home)
@@ -695,5 +700,5 @@ func readMetaFile(home string) (storeMeta, error) {
 //   - a < b (mod 2^63)
 //   - b > a after an integer rollover that is still within the distance of <2^63-1
 func isTxIDLessEqual(a, b uint64) bool {
-	return int64(a-b) <= 0
+	return a == b || a-b >= 1<<63
 }

libbeat/statestore/backend/memlog/store.go

@@ -63,7 +63,7 @@ type entry struct {
 func openStore(log *logp.Logger, home string, mode os.FileMode, bufSz uint, ignoreVersionCheck bool, checkpoint CheckpointPredicate) (*store, error) {
 	fi, err := os.Stat(home)
 	if os.IsNotExist(err) {
-		err = os.MkdirAll(home, os.ModeDir|0770)
+		err = os.MkdirAll(home, os.ModeDir|0o770)
 		if err != nil {
 			return nil, err
 		}
@@ -75,7 +75,16 @@ func openStore(log *logp.Logger, home string, mode os.FileMode, bufSz uint, igno
 	} else if !fi.Mode().IsDir() {
 		return nil, fmt.Errorf("'%v' is not a directory", home)
 	} else {
-		if err := pathEnsurePermissions(filepath.Join(home, metaFileName), mode); err != nil {
+		metaPath := filepath.Join(home, metaFileName)
+		if _, err := os.Stat(metaPath); os.IsNotExist(err) {
+			// Directory exists but meta.json is absent — the process likely crashed
+			// after MkdirAll but before writeMetaFile completed on a prior run.
+			if err := writeMetaFile(home, mode); err != nil {
+				return nil, err
+			}
+		} else if err != nil {
+			return nil, fmt.Errorf("failed to stat meta file: %w", err)
+		} else if err := pathEnsurePermissions(metaPath, mode); err != nil {
 			return nil, fmt.Errorf("failed to update meta file permissions: %w", err)
 		}
 	}

libbeat/statestore/backend/memlog/store_test.go

@@ -22,12 +22,75 @@ import (
 	"path/filepath"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/elastic/elastic-agent-libs/logp/logptest"
 	"github.com/elastic/elastic-agent-libs/mapstr"
 )
 
+func TestWriteMetaFileAtomic(t *testing.T) {
+	t.Run("writes_valid_meta_and_cleans_up_temp", func(t *testing.T) {
+		dir := t.TempDir()
+
+		require.NoError(t, writeMetaFile(dir, 0o600))
+
+		meta, err := readMetaFile(dir)
+		require.NoError(t, err)
+		require.NoError(t, checkMeta(meta))
+
+		matches, err := filepath.Glob(filepath.Join(dir, "*.tmp-*"))
+		require.NoError(t, err)
+		assert.Empty(t, matches, "no temp files should remain after successful write")
+	})
+
+	t.Run("no_partial_file_on_rotate_failure", func(t *testing.T) {
+		dir := t.TempDir()
+		// Place a directory at the meta.json path so SafeFileRotate (rename) fails.
+		// This simulates a crash-safe scenario: the destination is never touched
+		// until the temp file is fully written and synced.
+		metaPath := filepath.Join(dir, metaFileName)
+		require.NoError(t, os.Mkdir(metaPath, 0o700))
+
+		err := writeMetaFile(dir, 0o600)
+		require.Error(t, err)
+
+		// The directory at the destination must be intact (not replaced or emptied).
+		fi, statErr := os.Stat(metaPath)
+		require.NoError(t, statErr)
+		assert.True(t, fi.IsDir(), "destination should still be a directory after failed rotate")
+
+		// No temp files should remain.
+		matches, globErr := filepath.Glob(filepath.Join(dir, "*.tmp-*"))
+		require.NoError(t, globErr)
+		assert.Empty(t, matches, "no temp files should remain after failed rotate")
+	})
+}
+
+// TestOpenStoreRecoversMissingMeta verifies that openStore creates meta.json when
+// the store directory already exists but meta.json is absent — the scenario that
+// arises when the process crashes after MkdirAll but before writeMetaFile on a
+// prior run.
+func TestOpenStoreRecoversMissingMeta(t *testing.T) {
+	dir := t.TempDir()
+	// Simulate a crash after MkdirAll: the directory exists but is empty.
+	storePath := filepath.Join(dir, "store")
+	require.NoError(t, os.Mkdir(storePath, 0o770))
+
+	logger := logptest.NewTestingLogger(t, "")
+	store, err := openStore(logger.Named("test"), storePath, 0o660, 4096, false, func(_ uint64) bool {
+		return false
+	})
+	require.NoError(t, err, "openStore must succeed even when meta.json was missing")
+	store.Close()
+
+	require.FileExists(t, filepath.Join(storePath, metaFileName), "meta.json must be created on recovery")
+
+	meta, err := readMetaFile(storePath)
+	require.NoError(t, err)
+	assert.NoError(t, checkMeta(meta))
+}
+
 func TestRecoverFromCorruption(t *testing.T) {
 	path := t.TempDir()