Harden synchronization and membership safety

internal/cli/cli.go

@@ -1737,7 +1737,7 @@ func printHealthIssues(output io.Writer, issues []peer.HealthIssue, skip map[str
 }
 
 func (a *App) syncCommand() *cobra.Command {
-	var metadataOnly bool
+	var metadataOnly, dryRun bool
 	cmd := &cobra.Command{
 		Use: "sync", Args: cobra.NoArgs, Short: "Synchronize immediately",
 		RunE: func(cmd *cobra.Command, args []string) error {
@@ -1748,6 +1748,13 @@ func (a *App) syncCommand() *cobra.Command {
 			defer repo.Close()
 			ctx, cancel := commandContext(cmd)
 			defer cancel()
+			if dryRun {
+				plan, err := repo.SyncPlan(ctx, metadataOnly)
+				if err != nil {
+					return err
+				}
+				return a.writeSyncPlan(plan, metadataOnly)
+			}
 			if err := repo.Sync(ctx, true); err != nil {
 				return err
 			}
@@ -1773,9 +1780,62 @@ func (a *App) syncCommand() *cobra.Command {
 		},
 	}
 	cmd.Flags().BoolVar(&metadataOnly, "metadata-only", false, "skip pin refresh and quota enforcement")
+	cmd.Flags().BoolVar(&dryRun, "dry-run", false, "show the synchronization plan without applying it")
 	return cmd
 }
 
+func (a *App) writeSyncPlan(plan *repository.SyncPlan, metadataOnly bool) error {
+	mode := "full"
+	if metadataOnly {
+		mode = "metadata"
+	}
+	if a.output == "json" {
+		return writeJSON(a.Out, map[string]any{"action": "sync", "mode": mode, "dry_run": true, "plan": plan})
+	}
+	if a.quiet {
+		return nil
+	}
+	fmt.Fprintf(a.Out, "Dry run: sync (mode %s); no changes will be applied\n", mode)
+	fmt.Fprintf(a.Out, "Metadata: bidirectional sync with %d remote(s)\n", len(plan.Remotes))
+	for _, remote := range plan.Remotes {
+		fmt.Fprintf(a.Out, "  %s\t%s\n", remote.Name, remote.URL)
+	}
+	if metadataOnly {
+		fmt.Fprintln(a.Out, "Metadata mode skips pin refresh and cache enforcement")
+		return nil
+	}
+	if len(plan.Pins) == 0 {
+		fmt.Fprintln(a.Out, "Pins: none configured")
+	} else {
+		fmt.Fprintf(a.Out, "Pins: refresh %d pinned path(s)\n", len(plan.Pins))
+		for _, pin := range plan.Pins {
+			if pin.MissingFiles == 0 {
+				fmt.Fprintf(a.Out, "  %s\talready hydrated (%d file(s), %s)\n", displayPinnedPath(pin.Path), pin.LogicalFiles, config.FormatSize(pin.LogicalBytes))
+			} else {
+				fmt.Fprintf(a.Out, "  %s\twould fetch %d missing file(s), %s\n", displayPinnedPath(pin.Path), pin.MissingFiles, config.FormatSize(pin.MissingBytes))
+			}
+		}
+	}
+	fmt.Fprintf(a.Out, "Cache: %s / %s\n", config.FormatSize(plan.CacheBytes), config.FormatSize(plan.CacheLimit))
+	switch {
+	case plan.CacheBytes <= plan.CacheLimit:
+		fmt.Fprintln(a.Out, "  within limit; no eviction needed")
+	case len(plan.Evictions) == 0:
+		fmt.Fprintln(a.Out, "  over limit, but every cached file is pinned; nothing can be evicted")
+	default:
+		var bytes int64
+		for _, candidate := range plan.Evictions {
+			bytes += candidate.Size
+		}
+		fmt.Fprintf(a.Out, "  would evict %d file(s), %s, least recently used first:\n", len(plan.Evictions), config.FormatSize(bytes))
+		for _, candidate := range plan.Evictions {
+			fmt.Fprintf(a.Out, "    %s\t%s\n", candidate.Path, config.FormatSize(candidate.Size))
+		}
+		fmt.Fprintln(a.Out, "  inactive range-cache extents are reclaimed first; git-annex may keep candidates that lack another verified copy")
+	}
+	return nil
+}
+
 func (a *App) statusCommand() *cobra.Command {
 	return &cobra.Command{
 		Use: "status", Args: cobra.NoArgs, Short: "Show repository and cache status",

internal/cli/command_model.go

@@ -1027,7 +1027,6 @@ func (a *App) consolidatedSyncCommand() *cobra.Command {
 			return fmt.Errorf("unsupported sync mode %q; use metadata or full", mode)
 		}
 	}
-	a.addDryRun(command, "sync")
 	return command
 }
 
@@ -1090,6 +1089,14 @@ func (a *App) internalCommand() *cobra.Command {
 	pairing := &cobra.Command{Use: "pairing", Hidden: true}
 	pairing.AddCommand(detachChildren(completeParent, "complete")...)
 	command.AddCommand(pairing)
+	receiveGuard := &cobra.Command{Use: "receive-guard", Args: cobra.NoArgs, Hidden: true,
+		RunE: func(cmd *cobra.Command, _ []string) error {
+			if strings.TrimSpace(a.repo) == "" {
+				return errors.New("managed receive guard requires --repo")
+			}
+			return managed.ValidateReceiveUpdates(a.repo, cmd.InOrStdin())
+		}}
+	command.AddCommand(receiveGuard)
 	return command
 }
 

internal/cli/content_test.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
 
@@ -92,11 +93,13 @@ func TestContentListShowsLocalResidency(t *testing.T) {
 	}
 }
 
-func git(t *testing.T, ctx context.Context, directory string, args ...string) {
+func git(t *testing.T, ctx context.Context, directory string, args ...string) string {
 	t.Helper()
 	command := exec.CommandContext(ctx, "git", args...)
 	command.Dir = directory
-	if output, err := command.CombinedOutput(); err != nil {
+	output, err := command.CombinedOutput()
+	if err != nil {
 		t.Fatalf("git %s: %s: %v", args[0], output, err)
 	}
+	return strings.TrimSpace(string(output))
 }

internal/cli/sync_test.go

@@ -0,0 +1,108 @@
+package cli
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/bitbeamer/dfs/internal/repository"
+)
+
+func TestSyncDryRunShowsPlan(t *testing.T) {
+	if err := repository.CheckDependencies(); err != nil {
+		t.Skip(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	dir := t.TempDir()
+	t.Cleanup(func() {
+		// Frozen annex objects are read-only; let TempDir removal succeed.
+		_ = filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {
+			if err == nil {
+				_ = os.Chmod(path, info.Mode()|0o700)
+			}
+			return nil
+		})
+	})
+	repo, err := repository.Init(ctx, filepath.Join(dir, "repo"), "sync-test", 1<<20)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer repo.Close()
+	for name, content := range map[string]string{"small.txt": "small\n", "big.bin": "big\n"} {
+		if err := os.WriteFile(filepath.Join(repo.Config.Repository, name), []byte(content), 0o600); err != nil {
+			t.Fatal(err)
+		}
+	}
+	git(t, ctx, repo.Config.Repository, "annex", "add", "--", "small.txt", "big.bin")
+	if _, err := repo.CommitPending(ctx, "Add sync dry-run fixtures"); err != nil {
+		t.Fatal(err)
+	}
+	git(t, ctx, repo.Config.Repository, "annex", "drop", "--force", "--", "big.bin")
+	if err := repo.SetLocalPin("big.bin"); err != nil {
+		t.Fatal(err)
+	}
+
+	var output bytes.Buffer
+	app := &App{Out: &output, Err: &output, repo: repo.Config.Repository}
+	command := app.syncCommand()
+	command.SetArgs([]string{"--dry-run"})
+	if err := command.ExecuteContext(ctx); err != nil {
+		t.Fatal(err)
+	}
+	for _, expected := range []string{
+		"Dry run: sync (mode full); no changes will be applied",
+		"Metadata: bidirectional sync with 0 remote(s)",
+		"Pins: refresh 1 pinned path(s)",
+		"big.bin",
+		"would fetch 1 missing file(s)",
+		"within limit; no eviction needed",
+	} {
+		if !bytes.Contains(output.Bytes(), []byte(expected)) {
+			t.Fatalf("sync dry run missing %q: %q", expected, output.String())
+		}
+	}
+	// A dry run must not hydrate the missing pinned file.
+	if pinned, err := repo.Store.IsPinned("big.bin"); err != nil || !pinned {
+		t.Fatalf("pin after dry run: pinned=%v err=%v", pinned, err)
+	}
+	gitOutput := git(t, ctx, repo.Config.Repository, "annex", "find", "--in=here", "--", "big.bin")
+	if gitOutput != "" {
+		t.Fatalf("dry run hydrated big.bin: %q", gitOutput)
+	}
+
+	output.Reset()
+	command = app.syncCommand()
+	command.SetArgs([]string{"--dry-run", "--metadata-only"})
+	if err := command.ExecuteContext(ctx); err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Contains(output.Bytes(), []byte("Metadata mode skips pin refresh")) {
+		t.Fatalf("metadata dry run output = %q", output.String())
+	}
+
+	output.Reset()
+	jsonApp := &App{Out: &output, Err: &output, repo: repo.Config.Repository, output: "json"}
+	command = jsonApp.syncCommand()
+	command.SetArgs([]string{"--dry-run"})
+	if err := command.ExecuteContext(ctx); err != nil {
+		t.Fatal(err)
+	}
+	var envelope struct {
+		Action string `json:"action"`
+		Plan   struct {
+			CacheLimit int64            `json:"cache_limit_bytes"`
+			Pins       []map[string]any `json:"pins"`
+		} `json:"plan"`
+	}
+	if err := json.Unmarshal(output.Bytes(), &envelope); err != nil {
+		t.Fatalf("decode sync dry-run JSON: %v: %q", err, output.String())
+	}
+	if envelope.Action != "sync" || envelope.Plan.CacheLimit != 1<<20 || len(envelope.Plan.Pins) != 1 {
+		t.Fatalf("sync dry-run JSON = %+v", envelope)
+	}
+}

internal/core/content.go

@@ -248,6 +248,16 @@ func (s *Service) BeginWrite(ctx context.Context, request WriteRequest) (WriteTr
 		}
 		s.operationMu.Unlock()
 	}
+	writerUnlock, err := s.repo.BeginWriteGuard(ctx)
+	if err != nil {
+		return nil, classify("begin write", cleaned, err)
+	}
+	keepWriterLock := false
+	defer func() {
+		if !keepWriterLock {
+			writerUnlock()
+		}
+	}()
 	info, statErr := os.Stat(destination)
 	exists := statErr == nil
 	if errors.Is(statErr, os.ErrNotExist) {
@@ -328,8 +338,9 @@ func (s *Service) BeginWrite(ctx context.Context, request WriteRequest) (WriteTr
 			return cleanup(err)
 		}
 	}
+	keepWriterLock = true
 	return &writeTransaction{service: s, file: staging, destination: destination, path: cleaned,
-		operationID: request.OperationID, fingerprint: fingerprint}, nil
+		operationID: request.OperationID, fingerprint: fingerprint, writerUnlock: writerUnlock}, nil
 }
 
 func copyReaderAt(ctx context.Context, destination io.Writer, source io.ReaderAt, size int64) (int64, error) {
@@ -380,14 +391,15 @@ func syncDirectory(path string) error {
 }
 
 type writeTransaction struct {
-	service     *Service
-	file        *os.File
-	destination string
-	path        string
-	operationID string
-	fingerprint string
-	mu          sync.Mutex
-	finished    bool
+	service      *Service
+	file         *os.File
+	destination  string
+	path         string
+	operationID  string
+	fingerprint  string
+	writerUnlock func()
+	mu           sync.Mutex
+	finished     bool
 }
 
 func (t *writeTransaction) ReadAt(data []byte, offset int64) (int, error) {
@@ -530,15 +542,19 @@ func (t *writeTransaction) Commit(ctx context.Context) error {
 		return classify("commit", t.path, err)
 	}
 	t.finished = true
+	defer t.releaseWriterLock()
 	err := t.service.repo.WithWorkTreeLock(func() error {
 		if err := os.Rename(t.file.Name(), t.destination); err != nil {
 			return err
 		}
 		return syncDirectory(filepath.Dir(t.destination))
 	})
 	if err != nil {
-		_ = os.Remove(t.file.Name())
-		err = classify("commit", t.path, err)
+		publishErr := err
+		if quarantineErr := t.quarantine(); quarantineErr != nil {
+			publishErr = errors.Join(publishErr, fmt.Errorf("quarantine staged write: %w", quarantineErr))
+		}
+		err = classify("commit", t.path, publishErr)
 	} else if t.service.repo.Store != nil {
 		if info, statErr := os.Stat(t.destination); statErr != nil {
 			err = classify("commit metadata", t.path, statErr)
@@ -568,18 +584,41 @@ func (t *writeTransaction) Commit(ctx context.Context) error {
 	return err
 }
 
+func (t *writeTransaction) quarantine() error {
+	directory := filepath.Join(t.service.root, ".git", "dfs", "recovery", time.Now().UTC().Format("20060102T150405.000000000Z"), "writes")
+	if err := os.MkdirAll(directory, 0o700); err != nil {
+		return err
+	}
+	destination := filepath.Join(directory, filepath.Base(t.file.Name()))
+	if err := os.Rename(t.file.Name(), destination); err != nil {
+		return err
+	}
+	if err := syncDirectory(directory); err != nil {
+		return err
+	}
+	return syncDirectory(filepath.Dir(t.file.Name()))
+}
+
 func (t *writeTransaction) Abort(context.Context) error {
 	t.mu.Lock()
 	defer t.mu.Unlock()
 	if t.finished {
 		return nil
 	}
 	t.finished = true
+	defer t.releaseWriterLock()
 	err := errors.Join(t.file.Close(), os.Remove(t.file.Name()))
 	t.service.releasePending(t.operationID, t.fingerprint)
 	return classify("abort", t.path, err)
 }
 
+func (t *writeTransaction) releaseWriterLock() {
+	if t.writerUnlock != nil {
+		t.writerUnlock()
+		t.writerUnlock = nil
+	}
+}
+
 func (t *writeTransaction) Close() error { return t.Abort(context.Background()) }
 
 type replayWrite struct{ result error }

internal/core/core_test.go

@@ -165,6 +165,34 @@ func TestWriteCommitIsAtomicAndRetryIsIdempotent(t *testing.T) {
 	}
 }
 
+func TestFailedWritePublishQuarantinesStagedPayload(t *testing.T) {
+	service, root := testService(t, 16)
+	if err := os.Mkdir(filepath.Join(root, "parent"), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	transaction, err := service.BeginWrite(context.Background(), WriteRequest{Path: "parent/item.txt", Create: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := transaction.WriteAt([]byte("recover me"), 0); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Remove(filepath.Join(root, "parent")); err != nil {
+		t.Fatal(err)
+	}
+	if err := transaction.Commit(context.Background()); err == nil {
+		t.Fatal("commit unexpectedly succeeded after parent removal")
+	}
+	matches, err := filepath.Glob(filepath.Join(root, ".git", "dfs", "recovery", "*", "writes", "write-*"))
+	if err != nil || len(matches) != 1 {
+		t.Fatalf("quarantined writes = %q, %v", matches, err)
+	}
+	payload, err := os.ReadFile(matches[0])
+	if err != nil || string(payload) != "recover me" {
+		t.Fatalf("quarantined payload = %q, %v", payload, err)
+	}
+}
+
 func TestConcurrentIdempotentMutationExecutesOnce(t *testing.T) {
 	service, _ := testService(t, 16)
 	ctx := context.Background()