diff --git a/pf/schema_sidecar.go b/pf/schema_sidecar.go index e3aa556..42f83ed 100644 --- a/pf/schema_sidecar.go +++ b/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: diff --git a/pf/schema_sidecar_test.go b/pf/schema_sidecar_test.go index 44cd3e9..e4d9688 100644 --- a/pf/schema_sidecar_test.go +++ b/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)