Fix schema sidecar atomic write

pf/schema_sidecar.go

@@ -108,12 +108,55 @@ func writeSinkSchemaSidecarToDir(workDir string, schema SinkSchema) (string, err
 		return "", err
 	}
 
-	if err := os.WriteFile(sidecarPath, content, 0644); err != nil {
+	if err := writeFileAtomic(sidecarPath, content, 0644); err != nil {
 		return "", err
 	}
 	return sidecarPath, nil
 }
 
+func writeFileAtomic(path string, content []byte, perm os.FileMode) error {
+	var existingPerm os.FileMode
+	var hasExistingPerm bool
+	if info, err := os.Lstat(path); err == nil {
+		if info.Mode().IsRegular() {
+			existingPerm = info.Mode().Perm()
+			hasExistingPerm = true
+		}
+	} else if !os.IsNotExist(err) {
+		return err
+	}
+
+	tmpDir, err := os.MkdirTemp(filepath.Dir(path), filepath.Base(path)+".*.tmp")
+	if err != nil {
+		return err
+	}
+	defer func() {
+		_ = os.RemoveAll(tmpDir)
+	}()
+
+	tmpPath := filepath.Join(tmpDir, filepath.Base(path))
+	tmpFile, err := os.OpenFile(tmpPath, os.O_WRONLY|os.O_CREATE|os.O_EXCL, perm)
+	if err != nil {
+		return err
+	}
+
+	if _, err := tmpFile.Write(content); err != nil {
+		_ = tmpFile.Close()
+		return err
+	}
+	if hasExistingPerm {
+		if err := tmpFile.Chmod(existingPerm); err != nil {
+			_ = tmpFile.Close()
+			return err
+		}
+	}
+	if err := tmpFile.Close(); err != nil {
+		return err
+	}
+
+	return os.Rename(tmpPath, path)
+}
+
 func isNoSchemaType(schemaType string) bool {
 	switch strings.ToLower(strings.TrimSpace(schemaType)) {
 	case "", SchemaTypeBytes, SchemaTypeNone:

pf/schema_sidecar_test.go

@@ -23,6 +23,7 @@ import (
 	"encoding/json"
 	"os"
 	"path/filepath"
+	"runtime"
 	"testing"
 )
 
@@ -133,6 +134,76 @@ func TestWriteSinkSchemaSidecarToDirTrimsSchemaTypeAndPreservesCasing(t *testing
 	}
 }
 
+func TestWriteSinkSchemaSidecarToDirAtomicallyReplacesExistingSidecar(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("symlink permissions differ on Windows")
+	}
+	workDir := t.TempDir()
+	sidecarFile := filepath.Join(workDir, SinkSchemaSidecarFileName)
+	readOnlyTarget := filepath.Join(workDir, "read-only-target")
+	if err := os.WriteFile(readOnlyTarget, []byte("target"), 0400); err != nil {
+		t.Fatalf("write read-only target: %v", err)
+	}
+	defer func() {
+		_ = os.Chmod(readOnlyTarget, 0600)
+	}()
+	if err := os.Symlink(readOnlyTarget, sidecarFile); err != nil {
+		t.Fatalf("create sidecar symlink: %v", err)
+	}
+
+	sidecarPath, err := writeSinkSchemaSidecarToDir(workDir, SinkSchema{
+		SchemaType: SchemaTypeJSON,
+		SchemaData: `{"type":"record","name":"Student","fields":[]}`,
+	})
+	if err != nil {
+		t.Fatalf("writeSinkSchemaSidecarToDir returned error: %v", err)
+	}
+	if sidecarPath != sidecarFile {
+		t.Fatalf("sidecarPath = %q, want %q", sidecarPath, sidecarFile)
+	}
+	info, err := os.Lstat(sidecarFile)
+	if err != nil {
+		t.Fatalf("lstat sidecar: %v", err)
+	}
+	if info.Mode()&os.ModeSymlink != 0 {
+		t.Fatal("sidecar is still a symlink, want atomically replaced regular file")
+	}
+	targetContent, err := os.ReadFile(readOnlyTarget)
+	if err != nil {
+		t.Fatalf("read target: %v", err)
+	}
+	if string(targetContent) != "target" {
+		t.Fatalf("target content = %q, want unchanged", targetContent)
+	}
+}
+
+func TestWriteSinkSchemaSidecarToDirPreservesExistingSidecarPermissions(t *testing.T) {
+	for _, perm := range []os.FileMode{0000, 0600} {
+		t.Run(perm.String(), func(t *testing.T) {
+			workDir := t.TempDir()
+			sidecarFile := filepath.Join(workDir, SinkSchemaSidecarFileName)
+			if err := os.WriteFile(sidecarFile, []byte("old"), 0600); err != nil {
+				t.Fatalf("write existing sidecar: %v", err)
+			}
+			if err := os.Chmod(sidecarFile, perm); err != nil {
+				t.Fatalf("chmod existing sidecar: %v", err)
+			}
+
+			if _, err := writeSinkSchemaSidecarToDir(workDir, SinkSchema{SchemaType: SchemaTypeJSON}); err != nil {
+				t.Fatalf("writeSinkSchemaSidecarToDir returned error: %v", err)
+			}
+
+			info, err := os.Stat(sidecarFile)
+			if err != nil {
+				t.Fatalf("stat sidecar: %v", err)
+			}
+			if got := info.Mode().Perm(); got != perm {
+				t.Fatalf("sidecar permissions = %04o, want %04o", got, perm)
+			}
+		})
+	}
+}
+
 func TestSchemaTypeBoolUsesPulsarBooleanName(t *testing.T) {
 	if SchemaTypeBool != "boolean" {
 		t.Fatalf("SchemaTypeBool = %q, want boolean", SchemaTypeBool)