diff --git a/internal/cli/cli.go b/internal/cli/cli.go index ab80f7d..df51656 100644 --- a/internal/cli/cli.go +++ b/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", diff --git a/internal/cli/command_model.go b/internal/cli/command_model.go index 6b4b7e4..98f7670 100644 --- a/internal/cli/command_model.go +++ b/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 } diff --git a/internal/cli/content_test.go b/internal/cli/content_test.go index 2432fa5..e5eb482 100644 --- a/internal/cli/content_test.go +++ b/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)) } diff --git a/internal/cli/sync_test.go b/internal/cli/sync_test.go new file mode 100644 index 0000000..93b1085 --- /dev/null +++ b/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) + } +} diff --git a/internal/core/content.go b/internal/core/content.go index d012485..4105e77 100644 --- a/internal/core/content.go +++ b/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,6 +542,7 @@ 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 @@ -537,8 +550,11 @@ func (t *writeTransaction) Commit(ctx context.Context) error { 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,6 +584,21 @@ 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() @@ -575,11 +606,19 @@ func (t *writeTransaction) Abort(context.Context) error { 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 } diff --git a/internal/core/core_test.go b/internal/core/core_test.go index 62491a7..e8e8882 100644 --- a/internal/core/core_test.go +++ b/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() diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index af53b28..18d799a 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -74,7 +74,13 @@ func Run(repo *repository.Repository, options Options) (runErr error) { if receiveErr != nil { return } - if reason != "" { + switch reason { + case wakeup.WriterBegin: + scheduler.BeginWrite() + case wakeup.WriterEnd: + scheduler.EndWrite() + case "": + default: scheduler.Notify(reason) } } diff --git a/internal/managed/transport.go b/internal/managed/transport.go index 699d1fc..9ebb33f 100644 --- a/internal/managed/transport.go +++ b/internal/managed/transport.go @@ -39,6 +39,8 @@ const ALPN = "dfs-managed-v1" const PairALPN = "dfs-pair-v2" const managedDialTimeout = time.Second +const requestHeaderTimeout = 20 * time.Second +const requestHeaderLimit = 64 << 10 const ( contentAvailabilityBudget = 1500 * time.Millisecond @@ -77,6 +79,8 @@ type Server struct { done chan struct{} once sync.Once connections atomic.Int64 + hookOnce sync.Once + hookErr error } func Start(repo *repository.Repository, address string, diagnostic func(context.Context) ([]byte, error), pairingCertificate *tls.Certificate, pair func(context.Context, string, net.Addr, json.RawMessage) (json.RawMessage, error), pairClone func(context.Context, string, string) error, changed func(string, []string)) (*Server, error) { @@ -165,9 +169,8 @@ func (s *Server) serveConnection(connection *quic.Conn) { func (s *Server) serveStream(stream *quic.Stream, protocol string, remote net.Addr) { defer stream.Close() - reader := bufio.NewReaderSize(stream, 64<<10) - header, err := reader.ReadBytes('\n') - if err != nil || len(header) > 64<<10 { + header, reader, err := readRequestHeader(stream, requestHeaderTimeout) + if err != nil { writeResponse(stream, Response{Error: "read managed transport request"}) return } @@ -248,6 +251,26 @@ func (s *Server) serveStream(stream *quic.Stream, protocol string, remote net.Ad } } +type requestHeaderReader interface { + io.Reader + SetReadDeadline(time.Time) error +} + +func readRequestHeader(stream requestHeaderReader, timeout time.Duration) ([]byte, *bufio.Reader, error) { + if err := stream.SetReadDeadline(time.Now().Add(timeout)); err != nil { + return nil, nil, err + } + reader := bufio.NewReaderSize(stream, requestHeaderLimit+1) + header, err := reader.ReadSlice('\n') + if err != nil || len(header) > requestHeaderLimit { + return nil, nil, errors.New("managed transport request header exceeds its limit or deadline") + } + if err := stream.SetReadDeadline(time.Time{}); err != nil { + return nil, nil, err + } + return header, reader, nil +} + func (s *Server) serveHasContent(stream *quic.Stream, key string) { if key == "" || strings.ContainsAny(key, "\r\n\x00") { writeResponse(stream, Response{Error: "invalid annex key"}) @@ -486,7 +509,18 @@ func (s *Server) serveGit(stream *quic.Stream, input io.Reader, service string) treeBefore = worktreeTree(stream.Context(), s.repo.Config.Repository) pinRefBefore = gitRefValue(stream.Context(), s.repo.Config.Repository, membership.PinRef) } - command := exec.CommandContext(stream.Context(), service, s.repo.Config.Repository) + commandName := service + commandArgs := []string{s.repo.Config.Repository} + if service == "git-receive-pack" { + s.hookOnce.Do(func() { s.hookErr = installReceiveGuard(s.repo.Config.Repository) }) + if s.hookErr != nil { + writeResponse(stream, Response{Error: "prepare guarded Git receive"}) + return + } + commandName = "git" + commandArgs = []string{"-c", "core.hooksPath=" + filepath.Join(s.repo.Config.Repository, ".git", "dfs", "managed-hooks"), "receive-pack", s.repo.Config.Repository} + } + command := exec.CommandContext(stream.Context(), commandName, commandArgs...) processcommand.ConfigureCancellation(command) command.Stdin, command.Stdout, command.Stderr = input, stream, io.Discard if err := command.Start(); err != nil { @@ -520,6 +554,99 @@ func (s *Server) serveGit(stream *quic.Stream, input io.Reader, service string) } } +func installReceiveGuard(repositoryPath string) error { + directory := filepath.Join(repositoryPath, ".git", "dfs", "managed-hooks") + if err := os.MkdirAll(directory, 0o700); err != nil { + return err + } + executable, err := receiveGuardExecutable() + if err != nil { + return err + } + hook := "#!/bin/sh\nset -eu\nexec " + shellArgument(executable) + " --repo " + shellArgument(repositoryPath) + " internal receive-guard\n" + path := filepath.Join(directory, "pre-receive") + temporary, err := os.CreateTemp(directory, "pre-receive-*") + if err != nil { + return err + } + name := temporary.Name() + defer os.Remove(name) + if _, err := temporary.WriteString(hook); err != nil { + _ = temporary.Close() + return err + } + if err := temporary.Sync(); err != nil { + _ = temporary.Close() + return err + } + if err := temporary.Chmod(0o700); err != nil { + _ = temporary.Close() + return err + } + if err := temporary.Close(); err != nil { + return err + } + return os.Rename(name, path) +} + +var receiveGuardExecutable = os.Executable + +func shellArgument(value string) string { + return "'" + strings.ReplaceAll(value, "'", "'\"'\"'") + "'" +} + +// ValidateReceiveUpdates validates pre-receive input for managed control refs. +func ValidateReceiveUpdates(repositoryPath string, input io.Reader) error { + updates := make(map[string]membership.RefUpdate) + scanner := bufio.NewScanner(input) + for scanner.Scan() { + fields := strings.Fields(scanner.Text()) + if len(fields) != 3 { + return errors.New("invalid managed pre-receive update") + } + old, next, ref := fields[0], fields[1], fields[2] + if ref != membership.SharedRef && ref != membership.PinRef && ref != membership.ConfigRef { + continue + } + if zeroObjectID(next) { + return fmt.Errorf("DFS control ref %s cannot be deleted", ref) + } + if !zeroObjectID(old) { + command := exec.Command("git", "-C", repositoryPath, "merge-base", "--is-ancestor", old, next) + command.Env = managedReceiveGitEnvironment(os.Environ()) + if output, err := command.CombinedOutput(); err != nil { + return fmt.Errorf("DFS control ref %s cannot be rewound: %s", ref, strings.TrimSpace(string(output))) + } + } + updates[ref] = membership.RefUpdate{Old: old, New: next} + } + if err := scanner.Err(); err != nil { + return err + } + return membership.ValidateControlUpdates(repositoryPath, updates) +} + +func managedReceiveGitEnvironment(environment []string) []string { + result := make([]string, 0, len(environment)) + for _, value := range environment { + name := value + if index := strings.IndexByte(name, '='); index >= 0 { + name = name[:index] + } + switch name { + case "GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "GIT_PREFIX": + continue + default: + result = append(result, value) + } + } + return result +} + +func zeroObjectID(value string) bool { + return value != "" && strings.Trim(value, "0") == "" +} + func gitRefsValue(ctx context.Context, repositoryPath string) (string, bool) { output, err := exec.CommandContext(ctx, "git", "-C", repositoryPath, "for-each-ref", "--format=%(refname) %(objectname)", "refs").Output() if err != nil { diff --git a/internal/managed/transport_test.go b/internal/managed/transport_test.go index 4bdfd50..148786c 100644 --- a/internal/managed/transport_test.go +++ b/internal/managed/transport_test.go @@ -59,6 +59,15 @@ func TestMutuallyAuthenticatedQUICDiagnosticAndContent(t *testing.T) { } serverKey, serverRecord := managedTestRecord(t, serverRepo, filesystemID, "quic://127.0.0.1:1") clientKey, clientRecord := managedTestRecord(t, clientRepo, filesystemID, "quic://127.0.0.1:1") + clientRecord.Payload.Role = "member" + clientRecord, err = membership.Sign(clientRecord.Payload, clientKey) + if err != nil { + t.Fatal(err) + } + clientRecord, err = membership.Approve(clientRecord, serverRepo.Config.PeerID, serverKey) + if err != nil { + t.Fatal(err) + } for _, repo := range []*repository.Repository{serverRepo, clientRepo} { for _, record := range []membership.Record{serverRecord, clientRecord} { if err := membership.Save(repo.Config.Repository, record); err != nil { @@ -68,6 +77,9 @@ func TestMutuallyAuthenticatedQUICDiagnosticAndContent(t *testing.T) { t.Fatal(err) } } + if err := membership.TrustBootstrapAdmin(repo.Config.Repository, serverRecord); err != nil { + t.Fatal(err) + } } payload := []byte("managed content over quic\n") if err := os.WriteFile(filepath.Join(serverRepo.Config.Repository, "payload.txt"), payload, 0o644); err != nil { @@ -186,6 +198,9 @@ func TestMutuallyAuthenticatedQUICDiagnosticAndContent(t *testing.T) { if output, err := build.CombinedOutput(); err != nil { t.Fatalf("build DFS helper: %v\n%s", err, output) } + previousGuardExecutable := receiveGuardExecutable + receiveGuardExecutable = func() (string, error) { return binary, nil } + defer func() { receiveGuardExecutable = previousGuardExecutable }() if _, err := clientRepo.AdoptClonedPeer(ctx, serverRepo.Config.PeerID); err != nil { t.Fatalf("adopt cloned peer: %v", err) } @@ -193,6 +208,36 @@ func TestMutuallyAuthenticatedQUICDiagnosticAndContent(t *testing.T) { if err != nil { t.Fatal(err) } + clientRecord.Payload.Generation++ + clientRecord.Payload.UpdatedAt = time.Now().UTC().Add(time.Second) + clientRecord, err = membership.Sign(clientRecord.Payload, clientKey) + if err != nil { + t.Fatal(err) + } + clientRecord, err = membership.Approve(clientRecord, serverRepo.Config.PeerID, serverKey) + if err != nil { + t.Fatal(err) + } + if err := membership.Save(clientRepo.Config.Repository, clientRecord); err != nil { + t.Fatal(err) + } + if err := membership.Sync(ctx, clientRepo.Config.Repository, []string{remoteName}); err != nil { + t.Fatalf("authorized membership update over managed receive: %v", err) + } + controlPush := exec.CommandContext(ctx, "git", "-C", clientRepo.Config.Repository, "push", remoteName, + membership.SharedRef+":"+membership.SharedRef) + if output, err := controlPush.CombinedOutput(); err != nil { + t.Fatalf("push authorized membership update over managed receive: %v\n%s", err, output) + } + serverMembers, err := membership.LoadAll(serverRepo.Config.Repository) + if err != nil { + t.Fatal(err) + } + if !slices.ContainsFunc(serverMembers, func(record membership.Record) bool { + return record.Payload.PeerID == clientRecord.Payload.PeerID && record.Payload.Generation == clientRecord.Payload.Generation + }) { + t.Fatalf("authorized membership generation did not reach server: %#v", serverMembers) + } if err := clientRepo.Sync(ctx, true); err != nil { t.Fatalf("identify managed annex remote: %v", err) } @@ -447,6 +492,75 @@ func TestMutuallyAuthenticatedQUICDiagnosticAndContent(t *testing.T) { } } +func TestManagedReceiveGuardRejectsControlRefRewrites(t *testing.T) { + repositoryPath := t.TempDir() + run := func(input string, args ...string) ([]byte, error) { + command := exec.Command("git", args...) + command.Dir = repositoryPath + command.Stdin = strings.NewReader(input) + return command.CombinedOutput() + } + if output, err := run("", "init", "-b", "main"); err != nil { + t.Fatalf("init: %v\n%s", err, output) + } + if _, err := run("", "config", "user.name", "Guard Test"); err != nil { + t.Fatal(err) + } + if _, err := run("", "config", "user.email", "guard@example.invalid"); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(repositoryPath, "value"), []byte("one"), 0o644); err != nil { + t.Fatal(err) + } + if output, err := run("", "add", "value"); err != nil { + t.Fatalf("add: %v\n%s", err, output) + } + if output, err := run("", "commit", "-m", "one"); err != nil { + t.Fatalf("commit: %v\n%s", err, output) + } + oldBytes, _ := run("", "rev-parse", "HEAD") + if err := os.WriteFile(filepath.Join(repositoryPath, "value"), []byte("two"), 0o644); err != nil { + t.Fatal(err) + } + if _, err := run("", "commit", "-am", "two"); err != nil { + t.Fatal(err) + } + newBytes, _ := run("", "rev-parse", "HEAD") + oldID, newID := strings.TrimSpace(string(oldBytes)), strings.TrimSpace(string(newBytes)) + if err := ValidateReceiveUpdates(repositoryPath, strings.NewReader(newID+" "+oldID+" refs/heads/dfs-membership\n")); err == nil || !strings.Contains(err.Error(), "cannot be rewound") { + t.Fatalf("rewind result = %v", err) + } + if err := ValidateReceiveUpdates(repositoryPath, strings.NewReader(newID+" "+strings.Repeat("0", 40)+" refs/heads/dfs-pins\n")); err == nil || !strings.Contains(err.Error(), "cannot be deleted") { + t.Fatalf("delete result = %v", err) + } +} + +func TestRequestHeaderIsBoundedAndDeadlineLimited(t *testing.T) { + server, client := net.Pipe() + writeDone := make(chan struct{}) + go func() { + _, _ = client.Write(bytes.Repeat([]byte{'x'}, requestHeaderLimit+1)) + _ = client.Close() + close(writeDone) + }() + if _, _, err := readRequestHeader(server, time.Second); err == nil { + t.Fatal("oversized request header was accepted") + } + _ = server.Close() + <-writeDone + + server, client = net.Pipe() + defer server.Close() + defer client.Close() + started := time.Now() + if _, _, err := readRequestHeader(server, 25*time.Millisecond); err == nil { + t.Fatal("stalled request header was accepted") + } + if elapsed := time.Since(started); elapsed > time.Second { + t.Fatalf("stalled request header exceeded its deadline: %v", elapsed) + } +} + func TestLocalIPv4PrefersIPv4ForLocalHostname(t *testing.T) { lookupCalled := false lookup := func(_ context.Context, network, hostname string) ([]net.IP, error) { diff --git a/internal/membership/membership.go b/internal/membership/membership.go index 1538f9b..366c8b1 100644 --- a/internal/membership/membership.go +++ b/internal/membership/membership.go @@ -29,6 +29,7 @@ const ( ConfigRef = "refs/heads/dfs-config" privateKeyFile = "membership-key.pem" trustedMembersFile = "trusted-members.json" + bootstrapAdminsFile = "bootstrap-admins.json" revokedMembersFile = "revoked-members.json" membersPrefix = "members/" revocationsPrefix = "revocations/" @@ -233,10 +234,14 @@ func SetPinPolicy(repositoryPath, filesystemID, peerID, path string, pinned bool } func LoadPinPolicies(repositoryPath, filesystemID string) ([]PinPolicy, error) { - trusted, err := LoadTrusted(repositoryPath) + records, err := Accepted(repositoryPath, filesystemID) if err != nil { return nil, err } + trusted := make(map[string]string, len(records)) + for _, record := range records { + trusted[record.Payload.PeerID] = record.Payload.SigningPublicKey + } files, err := loadSharedPrefix(repositoryPath, PinRef, pinsPrefix) if err != nil { return nil, err @@ -557,6 +562,7 @@ func TrustStateVersion(repositoryPath string) (string, error) { filepath.Join(repositoryPath, ".git", "refs", "heads", "dfs-membership"), filepath.Join(repositoryPath, ".git", "packed-refs"), filepath.Join(repositoryPath, filepath.FromSlash(config.Directory), trustedMembersFile), + filepath.Join(repositoryPath, filepath.FromSlash(config.Directory), bootstrapAdminsFile), filepath.Join(repositoryPath, filepath.FromSlash(config.Directory), revokedMembersFile), } hash := sha256.New() @@ -588,8 +594,8 @@ func IsRevoked(repositoryPath, peerID string) (bool, error) { } // Accepted returns the non-revoked records whose approval chain reaches a -// locally trusted peer. Newly accepted peers are persisted as trust anchors so -// offline reconciliation does not depend on record ordering. +// locally pinned, self-approved bootstrap administrator. Every subsequent +// admission and role change must be approved by an accepted administrator. func Accepted(repositoryPath, filesystemID string, legacyPeerIDs ...string) ([]Record, error) { for _, peerID := range legacyPeerIDs { if err := Trust(repositoryPath, peerID, ""); err != nil { @@ -610,7 +616,6 @@ func Accepted(repositoryPath, filesystemID string, legacyPeerIDs ...string) ([]R byID[record.Payload.PeerID] = record } } - changed := true for id, key := range trusted { if key == "" { if record, ok := byID[id]; ok { @@ -618,24 +623,32 @@ func Accepted(repositoryPath, filesystemID string, legacyPeerIDs ...string) ([]R } } } - for changed { - changed = false + bootstrap, err := loadBootstrapAdmins(repositoryPath) + if err != nil { + return nil, err + } + if len(bootstrap) == 0 { + // One-time migration for repositories created before bootstrap roots + // were stored separately. Only a pinned generation-one administrator + // with a valid self-approval can seed the authority chain. for id, record := range byID { - if trusted[id] != "" || trusted[record.ApprovedBy] == "" { - continue + if trusted[id] == record.Payload.SigningPublicKey && record.Payload.Generation == 1 && + record.Payload.Role == "admin" && record.ApprovedBy == id && VerifyApproval(record, record) == nil { + bootstrap[id] = record.Payload.SigningPublicKey } - approver, ok := byID[record.ApprovedBy] - if ok && approver.Payload.SigningPublicKey == trusted[record.ApprovedBy] && VerifyApproval(record, approver) == nil { - trusted[id] = record.Payload.SigningPublicKey - changed = true + } + if len(bootstrap) > 0 { + if err := saveBootstrapAdmins(repositoryPath, bootstrap); err != nil { + return nil, err } } } - revoked, err := acceptedRevocations(repositoryPath, filesystemID, byID, trusted) + persistedRevoked, err := loadRevoked(repositoryPath) if err != nil { return nil, err } - persistedRevoked, err := loadRevoked(repositoryPath) + authorized := authorizeRecords(byID, bootstrap, persistedRevoked) + revoked, err := acceptedRevocations(repositoryPath, filesystemID, byID, authorized) if err != nil { return nil, err } @@ -652,13 +665,14 @@ func Accepted(repositoryPath, filesystemID string, legacyPeerIDs ...string) ([]R } } revoked = persistedRevoked + authorized = authorizeRecords(byID, bootstrap, revoked) var accepted []Record - for id, key := range trusted { + for id, key := range authorized { if record, ok := byID[id]; ok && record.Payload.SigningPublicKey == key && !record.Payload.Revoked && !revoked[id] { accepted = append(accepted, record) } } - for id, key := range trusted { + for id, key := range authorized { if key != "" { if err := Trust(repositoryPath, id, key); err != nil { return nil, err @@ -669,6 +683,88 @@ func Accepted(repositoryPath, filesystemID string, legacyPeerIDs ...string) ([]R return accepted, nil } +// TrustBootstrapAdmin pins an administrator delivered through an authenticated +// bootstrap path, such as initial filesystem creation or pairing. The local pin +// is the root of the later signed approval chain. +func TrustBootstrapAdmin(repositoryPath string, record Record) error { + if err := VerifySelf(record); err != nil { + return err + } + if record.Payload.Role != "admin" { + return errors.New("bootstrap membership is not an administrator") + } + if err := Trust(repositoryPath, record.Payload.PeerID, record.Payload.SigningPublicKey); err != nil { + return err + } + bootstrap, err := loadBootstrapAdmins(repositoryPath) + if err != nil { + return err + } + if key := bootstrap[record.Payload.PeerID]; key != "" && key != record.Payload.SigningPublicKey { + return errors.New("bootstrap administrator key replacement was rejected") + } + bootstrap[record.Payload.PeerID] = record.Payload.SigningPublicKey + return saveBootstrapAdmins(repositoryPath, bootstrap) +} + +func authorizeRecords(byID map[string]Record, bootstrap map[string]string, revoked map[string]bool) map[string]string { + authorized := make(map[string]string) + for id, key := range bootstrap { + record, ok := byID[id] + if !ok || key == "" || record.Payload.SigningPublicKey != key || record.Payload.Role != "admin" || + revoked[id] || VerifySelf(record) != nil { + continue + } + authorized[id] = key + } + for changed := true; changed; { + changed = false + for id, record := range byID { + if authorized[id] != "" || revoked[id] || revoked[record.ApprovedBy] { + continue + } + approver, ok := byID[record.ApprovedBy] + if !ok || approver.Payload.Role != "admin" || approver.Payload.SigningPublicKey != authorized[record.ApprovedBy] { + continue + } + if VerifyApproval(record, approver) == nil { + authorized[id] = record.Payload.SigningPublicKey + changed = true + } + } + } + return authorized +} + +func loadBootstrapAdmins(repositoryPath string) (map[string]string, error) { + path := filepath.Join(repositoryPath, filepath.FromSlash(config.Directory), bootstrapAdminsFile) + data, err := os.ReadFile(path) + if errors.Is(err, os.ErrNotExist) { + return make(map[string]string), nil + } + if err != nil { + return nil, err + } + result := make(map[string]string) + if err := json.Unmarshal(data, &result); err != nil { + return nil, err + } + return result, nil +} + +func saveBootstrapAdmins(repositoryPath string, roots map[string]string) error { + data, err := json.MarshalIndent(roots, "", " ") + if err != nil { + return err + } + data = append(data, '\n') + path := filepath.Join(repositoryPath, filepath.FromSlash(config.Directory), bootstrapAdminsFile) + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + return err + } + return replacePrivateFile(path, data) +} + // CurrentMachines collapses accepted records left behind by peer reinstalls to // the newest identity for each named machine. A reinstalled host keeps its // hostname and peer name but receives a new peer ID and signing key. @@ -704,39 +800,10 @@ func CurrentMachines(records []Record) ([]Record, map[string]bool) { } func AcceptedRevocations(repositoryPath, filesystemID string) (map[string]bool, error) { - trusted, err := LoadTrusted(repositoryPath) - if err != nil { - return nil, err - } - records, err := LoadAll(repositoryPath) - if err != nil { - return nil, err - } - byID := make(map[string]Record) - for _, record := range records { - byID[record.Payload.PeerID] = record - } - revoked, err := acceptedRevocations(repositoryPath, filesystemID, byID, trusted) - if err != nil { + if _, err := Accepted(repositoryPath, filesystemID); err != nil { return nil, err } - persisted, err := loadRevoked(repositoryPath) - if err != nil { - return nil, err - } - changed := false - for id := range revoked { - if !persisted[id] { - persisted[id] = true - changed = true - } - } - if changed { - if err := saveRevoked(repositoryPath, persisted); err != nil { - return nil, err - } - } - return persisted, nil + return loadRevoked(repositoryPath) } func canonicalPayload(payload Payload) ([]byte, error) { return json.Marshal(payload) } @@ -833,11 +900,12 @@ func validateRevocation(revocation Revocation) error { } func acceptedRevocations(repositoryPath, filesystemID string, records map[string]Record, trusted map[string]string) (map[string]bool, error) { - result := make(map[string]bool) files, err := loadSharedPrefix(repositoryPath, SharedRef, revocationsPrefix) if err != nil { return nil, err } + var candidates []Revocation + revokedAt := make(map[string]time.Time) for path, data := range files { name := strings.TrimPrefix(path, revocationsPrefix) var revocation Revocation @@ -846,6 +914,18 @@ func acceptedRevocations(repositoryPath, filesystemID string, records map[string } approver, ok := records[revocation.RevokedBy] if ok && approver.Payload.SigningPublicKey == trusted[revocation.RevokedBy] && VerifyRevocation(revocation, approver) == nil { + candidates = append(candidates, revocation) + if current, found := revokedAt[revocation.PeerID]; !found || revocation.UpdatedAt.Before(current) { + revokedAt[revocation.PeerID] = revocation.UpdatedAt + } + } + } + result := make(map[string]bool) + for _, revocation := range candidates { + // Preserve revocations issued before an administrator was removed, but + // reject claims issued at or after that administrator's revocation. + issuerRevokedAt, issuerRevoked := revokedAt[revocation.RevokedBy] + if !issuerRevoked || revocation.UpdatedAt.Before(issuerRevokedAt) { result[revocation.PeerID] = true } } @@ -1095,7 +1175,33 @@ func Sync(ctx context.Context, repositoryPath string, remotes []string) error { if err := syncFilesystemConfig(ctx, repositoryPath, remotes); err != nil { return err } - return syncPinPolicies(ctx, repositoryPath, remotes, trusted) + pinSigners, err := acceptedSigningKeys(repositoryPath) + if err != nil { + return err + } + return syncPinPolicies(ctx, repositoryPath, remotes, pinSigners) +} + +func acceptedSigningKeys(repositoryPath string) (map[string]string, error) { + records, err := LoadAll(repositoryPath) + if err != nil { + return nil, err + } + filesystems := make(map[string]bool) + for _, record := range records { + filesystems[record.Payload.FileSystemID] = true + } + result := make(map[string]string) + for filesystemID := range filesystems { + accepted, err := Accepted(repositoryPath, filesystemID) + if err != nil { + return nil, err + } + for _, record := range accepted { + result[record.Payload.PeerID] = record.Payload.SigningPublicKey + } + } + return result, nil } func syncFilesystemConfig(ctx context.Context, repositoryPath string, remotes []string) error { @@ -1396,6 +1502,202 @@ func validPinDocument(path string, data []byte, trusted map[string]string) bool return verifyPinPolicy(policy, trusted[policy.IssuedBy]) == nil } +// RefUpdate describes one proposed managed-transport control-ref update. +type RefUpdate struct { + Old string + New string +} + +// ValidateControlUpdates validates the complete proposed control-plane state +// before receive-pack updates any refs. It accepts legitimate generation +// updates while rejecting deletion, rollback, invalid signatures, and records +// whose authorization chain does not reach a locally pinned administrator. +func ValidateControlUpdates(repositoryPath string, updates map[string]RefUpdate) error { + if len(updates) == 0 { + return nil + } + membershipRef := SharedRef + previousMembershipRef := SharedRef + if update, ok := updates[SharedRef]; ok { + membershipRef = update.New + previousMembershipRef = update.Old + } + authorized, revoked, err := authorizedRecordsAtRef(repositoryPath, membershipRef, previousMembershipRef) + if err != nil { + return fmt.Errorf("validate proposed membership: %w", err) + } + for ref, update := range updates { + if ref != SharedRef && ref != PinRef && ref != ConfigRef { + continue + } + if err := validateDocumentContinuity(repositoryPath, ref, update.Old, update.New); err != nil { + return err + } + switch ref { + case SharedRef: + // authorizedRecordsAtRef validated every proposed document. + case PinRef: + files, err := loadSharedPrefix(repositoryPath, update.New, "") + if err != nil { + return err + } + for path, data := range files { + var policy PinPolicy + if json.Unmarshal(data, &policy) != nil || path != pinPolicyPath(policy.Path) || revoked[policy.IssuedBy] { + return fmt.Errorf("invalid cluster pin document %s", path) + } + record, ok := authorized[policy.IssuedBy] + if !ok || policy.FileSystemID != record.Payload.FileSystemID || verifyPinPolicy(policy, record.Payload.SigningPublicKey) != nil { + return fmt.Errorf("cluster pin document %s was not signed by an accepted member", path) + } + } + case ConfigRef: + files, err := loadSharedPrefix(repositoryPath, update.New, "") + if err != nil { + return err + } + for path, data := range files { + if path != filesystemConfigPath { + return fmt.Errorf("unexpected filesystem configuration document %s", path) + } + var value FilesystemConfig + if json.Unmarshal(data, &value) != nil { + return errors.New("invalid filesystem configuration document") + } + record, ok := authorized[value.IssuedBy] + if !ok || record.Payload.Role != "admin" || value.FileSystemID != record.Payload.FileSystemID || revoked[value.IssuedBy] || + verifyFilesystemConfig(value, record.Payload.SigningPublicKey) != nil { + return errors.New("filesystem configuration was not signed by an accepted administrator") + } + } + } + } + return nil +} + +func authorizedRecordsAtRef(repositoryPath, ref, previousRef string) (map[string]Record, map[string]bool, error) { + files, err := loadSharedPrefix(repositoryPath, ref, "") + if err != nil { + return nil, nil, err + } + byID := make(map[string]Record) + filesystemID := "" + for path, data := range files { + if strings.HasPrefix(path, revocationsPrefix) { + continue + } + if !strings.HasPrefix(path, membersPrefix) { + return nil, nil, fmt.Errorf("unexpected membership document %s", path) + } + var record Record + if json.Unmarshal(data, &record) != nil || VerifySelf(record) != nil || path != membersPrefix+record.Payload.PeerID+".json" { + return nil, nil, fmt.Errorf("invalid membership document %s", path) + } + if filesystemID != "" && filesystemID != record.Payload.FileSystemID { + return nil, nil, errors.New("membership ref contains multiple filesystem identities") + } + filesystemID = record.Payload.FileSystemID + byID[record.Payload.PeerID] = record + } + bootstrap, err := loadBootstrapAdmins(repositoryPath) + if err != nil { + return nil, nil, err + } + persisted, err := loadRevoked(repositoryPath) + if err != nil { + return nil, nil, err + } + preRevocation := authorizeRecords(byID, bootstrap, persisted) + revoked := make(map[string]bool, len(persisted)) + for id := range persisted { + revoked[id] = true + } + var proposed []Revocation + previousFiles := make(map[string][]byte) + if previousRef != "" && strings.Trim(previousRef, "0") != "" { + previousFiles, err = loadSharedPrefix(repositoryPath, previousRef, revocationsPrefix) + if err != nil { + return nil, nil, err + } + } + for path, data := range files { + if !strings.HasPrefix(path, revocationsPrefix) { + continue + } + var value Revocation + if json.Unmarshal(data, &value) != nil || path != revocationsPrefix+value.PeerID+".json" { + return nil, nil, fmt.Errorf("invalid revocation document %s", path) + } + if bytes.Equal(previousFiles[path], data) { + revoked[value.PeerID] = true + continue + } + approver, ok := byID[value.RevokedBy] + if !ok || preRevocation[value.RevokedBy] != approver.Payload.SigningPublicKey || + approver.Payload.Role != "admin" || VerifyRevocation(value, approver) != nil { + return nil, nil, fmt.Errorf("revocation of %s was not signed by an accepted administrator", value.PeerID) + } + proposed = append(proposed, value) + revoked[value.PeerID] = true + } + for _, value := range proposed { + if revoked[value.RevokedBy] { + return nil, nil, fmt.Errorf("revoked administrator %s cannot issue revocations", value.RevokedBy) + } + } + authorizedKeys := authorizeRecords(byID, bootstrap, revoked) + authorized := make(map[string]Record, len(authorizedKeys)) + for id, record := range byID { + if revoked[id] { + continue + } + if authorizedKeys[id] != record.Payload.SigningPublicKey { + return nil, nil, fmt.Errorf("membership record %s has no accepted administrator approval", id) + } + authorized[id] = record + } + return authorized, revoked, nil +} + +func validateDocumentContinuity(repositoryPath, ref, old, new string) error { + if strings.TrimSpace(new) == "" || strings.Trim(new, "0") == "" { + return fmt.Errorf("DFS control ref %s cannot be deleted", ref) + } + if strings.TrimSpace(old) == "" || strings.Trim(old, "0") == "" { + return nil + } + before, err := loadSharedPrefix(repositoryPath, old, "") + if err != nil { + return err + } + after, err := loadSharedPrefix(repositoryPath, new, "") + if err != nil { + return err + } + for path, previous := range before { + candidate, ok := after[path] + if !ok { + return fmt.Errorf("DFS control document %s cannot be deleted", path) + } + if !bytes.Equal(previous, candidate) && !controlDocumentAdvances(ref, path, candidate, previous) { + return fmt.Errorf("DFS control document %s did not advance its generation", path) + } + } + return nil +} + +func controlDocumentAdvances(ref, path string, candidate, previous []byte) bool { + if ref != ConfigRef { + return newerSharedDocument(path, candidate, previous) + } + var next, current FilesystemConfig + if json.Unmarshal(candidate, &next) != nil || json.Unmarshal(previous, ¤t) != nil { + return false + } + return next.Generation > current.Generation || + (next.Generation == current.Generation && next.UpdatedAt.After(current.UpdatedAt)) +} + func newerSharedDocument(path string, candidate, current []byte) bool { var candidateGeneration, currentGeneration uint64 var candidateTime, currentTime time.Time @@ -1480,7 +1782,7 @@ func runGit(ctx context.Context, repositoryPath string, input []byte, arguments func runGitEnv(ctx context.Context, repositoryPath string, environment []string, input []byte, arguments ...string) ([]byte, error) { command := exec.CommandContext(ctx, "git", append([]string{"-C", repositoryPath}, arguments...)...) - command.Env = append(os.Environ(), environment...) + command.Env = append(cleanGitRepositoryEnvironment(os.Environ()), environment...) command.Env = append(command.Env, "GIT_AUTHOR_NAME=DFS", "GIT_AUTHOR_EMAIL=dfs@localhost", "GIT_COMMITTER_NAME=DFS", "GIT_COMMITTER_EMAIL=dfs@localhost", @@ -1492,3 +1794,20 @@ func runGitEnv(ctx context.Context, repositoryPath string, environment []string, } return output, nil } + +func cleanGitRepositoryEnvironment(environment []string) []string { + result := make([]string, 0, len(environment)) + for _, value := range environment { + name := value + if index := strings.IndexByte(name, '='); index >= 0 { + name = name[:index] + } + switch name { + case "GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "GIT_PREFIX": + continue + default: + result = append(result, value) + } + } + return result +} diff --git a/internal/membership/membership_test.go b/internal/membership/membership_test.go index 09da419..f0657eb 100644 --- a/internal/membership/membership_test.go +++ b/internal/membership/membership_test.go @@ -69,6 +69,16 @@ func TestAcceptedFollowsSignedApprovalChain(t *testing.T) { if err := SaveRevocation(shared, revocation); err != nil { t.Fatal(err) } + policy := PinPolicy{Version: Version, FileSystemID: filesystemID, Path: "shared", Pinned: true, + Generation: 1, UpdatedAt: time.Now().UTC(), IssuedBy: "bob-peer"} + policy.Signature = base64.RawStdEncoding.EncodeToString(ed25519.Sign(bobKey, pinPolicyBytes(policy))) + data, err := json.Marshal(policy) + if err != nil { + t.Fatal(err) + } + if err := writePinPolicyFile(shared, pinPolicyPath(policy.Path), data); err != nil { + t.Fatal(err) + } accepted, err = Accepted(shared, filesystemID) if err != nil { t.Fatal(err) @@ -76,6 +86,16 @@ func TestAcceptedFollowsSignedApprovalChain(t *testing.T) { if len(accepted) != 1 || accepted[0].Payload.PeerID != "alice-peer" { t.Fatalf("membership after revocation = %#v", accepted) } + if policies, err := LoadPinPolicies(shared, filesystemID); err != nil || len(policies) != 0 { + t.Fatalf("revoked issuer pin policies = %#v, %v", policies, err) + } + signers, err := acceptedSigningKeys(shared) + if err != nil { + t.Fatal(err) + } + if validPinDocument(pinPolicyPath(policy.Path), data, signers) { + t.Fatal("revoked issuer retained pin synchronization authority") + } if err := removeSharedFile(shared, revocationsPrefix+"bob-peer.json"); err != nil { t.Fatal(err) } @@ -85,6 +105,201 @@ func TestAcceptedFollowsSignedApprovalChain(t *testing.T) { } } +func TestAcceptedRequiresAdministratorForAdmissionsAndRoleChanges(t *testing.T) { + repositoryPath := gitMembershipRepository(t) + filesystemID := strings.Repeat("b", 40) + adminKey, adminPublic, _ := EnsureKey(t.TempDir()) + memberKey, memberPublic, _ := EnsureKey(t.TempDir()) + joinerKey, joinerPublic, _ := EnsureKey(t.TempDir()) + + admin := signedTestRecord(t, filesystemID, "admin-peer", "admin", adminPublic, adminKey) + admin, _ = Approve(admin, "admin-peer", adminKey) + member := signedTestRecord(t, filesystemID, "member-peer", "member", memberPublic, memberKey) + member, _ = Approve(member, "admin-peer", adminKey) + joiner := signedTestRecord(t, filesystemID, "joiner-peer", "member", joinerPublic, joinerKey) + joiner, _ = Approve(joiner, "member-peer", memberKey) + for _, record := range []Record{admin, member, joiner} { + if err := Save(repositoryPath, record); err != nil { + t.Fatal(err) + } + } + if err := Trust(repositoryPath, admin.Payload.PeerID, adminPublic); err != nil { + t.Fatal(err) + } + accepted, err := Accepted(repositoryPath, filesystemID) + if err != nil { + t.Fatal(err) + } + if len(accepted) != 2 { + t.Fatalf("non-admin admission accepted: %#v", accepted) + } + + escalated := signedTestRecord(t, filesystemID, "member-peer", "admin", memberPublic, memberKey) + escalated.Payload.Generation = 2 + escalated.Payload.UpdatedAt = time.Now().UTC().Add(time.Second) + escalated, _ = Sign(escalated.Payload, memberKey) + escalated, _ = Approve(escalated, "member-peer", memberKey) + if err := Save(repositoryPath, escalated); err != nil { + t.Fatal(err) + } + accepted, err = Accepted(repositoryPath, filesystemID) + if err != nil { + t.Fatal(err) + } + if len(accepted) != 1 || accepted[0].Payload.PeerID != "admin-peer" { + t.Fatalf("self-approved role escalation accepted: %#v", accepted) + } +} + +func TestAcceptedRevocationsRejectsTrustedSelfEscalation(t *testing.T) { + repositoryPath := gitMembershipRepository(t) + filesystemID := strings.Repeat("d", 40) + rootKey, rootPublic, _ := EnsureKey(t.TempDir()) + memberKey, memberPublic, _ := EnsureKey(t.TempDir()) + root := signedTestRecord(t, filesystemID, "root-peer", "admin", rootPublic, rootKey) + root, _ = Approve(root, "root-peer", rootKey) + member := signedTestRecord(t, filesystemID, "member-peer", "member", memberPublic, memberKey) + member, _ = Approve(member, "root-peer", rootKey) + for _, record := range []Record{root, member} { + if err := Save(repositoryPath, record); err != nil { + t.Fatal(err) + } + } + if err := TrustBootstrapAdmin(repositoryPath, root); err != nil { + t.Fatal(err) + } + if _, err := Accepted(repositoryPath, filesystemID); err != nil { + t.Fatal(err) + } + escalated := member + escalated.Payload.Role = "admin" + escalated.Payload.Generation++ + escalated.Payload.UpdatedAt = time.Now().UTC().Add(time.Second) + escalated, _ = Sign(escalated.Payload, memberKey) + escalated, _ = Approve(escalated, escalated.Payload.PeerID, memberKey) + if err := Save(repositoryPath, escalated); err != nil { + t.Fatal(err) + } + revocation, _ := Revoke(filesystemID, root.Payload.PeerID, escalated.Payload.PeerID, 2, memberKey) + if err := SaveRevocation(repositoryPath, revocation); err != nil { + t.Fatal(err) + } + revoked, err := AcceptedRevocations(repositoryPath, filesystemID) + if err != nil { + t.Fatal(err) + } + if revoked[root.Payload.PeerID] { + t.Fatal("trusted member used self-escalation to issue a sticky revocation") + } +} + +func TestControlUpdateAllowsAuthorizedGenerationAndRejectsEscalationAndDeletion(t *testing.T) { + repositoryPath := gitMembershipRepository(t) + filesystemID := strings.Repeat("e", 40) + rootKey, rootPublic, _ := EnsureKey(t.TempDir()) + memberKey, memberPublic, _ := EnsureKey(t.TempDir()) + root := signedTestRecord(t, filesystemID, "root-peer", "admin", rootPublic, rootKey) + root, _ = Approve(root, "root-peer", rootKey) + member := signedTestRecord(t, filesystemID, "member-peer", "member", memberPublic, memberKey) + member, _ = Approve(member, "root-peer", rootKey) + for _, record := range []Record{root, member} { + if err := Save(repositoryPath, record); err != nil { + t.Fatal(err) + } + } + if err := TrustBootstrapAdmin(repositoryPath, root); err != nil { + t.Fatal(err) + } + old := strings.TrimSpace(gitMembershipRunOutput(t, repositoryPath, "rev-parse", SharedRef)) + root.Payload.Generation++ + root.Payload.UpdatedAt = time.Now().UTC().Add(time.Second) + root.Payload.QUICEndpoint = "quic://root.local:9000" + root, _ = Sign(root.Payload, rootKey) + root, _ = Approve(root, root.Payload.PeerID, rootKey) + if err := Save(repositoryPath, root); err != nil { + t.Fatal(err) + } + next := strings.TrimSpace(gitMembershipRunOutput(t, repositoryPath, "rev-parse", SharedRef)) + if err := ValidateControlUpdates(repositoryPath, map[string]RefUpdate{SharedRef: {Old: old, New: next}}); err != nil { + t.Fatalf("authorized membership generation was rejected: %v", err) + } + + beforeEscalation := next + escalated := member + escalated.Payload.Role = "admin" + escalated.Payload.Generation++ + escalated.Payload.UpdatedAt = time.Now().UTC().Add(2 * time.Second) + escalated, _ = Sign(escalated.Payload, memberKey) + escalated, _ = Approve(escalated, escalated.Payload.PeerID, memberKey) + if err := Save(repositoryPath, escalated); err != nil { + t.Fatal(err) + } + afterEscalation := strings.TrimSpace(gitMembershipRunOutput(t, repositoryPath, "rev-parse", SharedRef)) + if err := ValidateControlUpdates(repositoryPath, map[string]RefUpdate{SharedRef: {Old: beforeEscalation, New: afterEscalation}}); err == nil { + t.Fatal("managed receive accepted self-approved administrator escalation") + } + + member.Payload.Generation = 3 + member.Payload.UpdatedAt = time.Now().UTC().Add(3 * time.Second) + member, _ = Sign(member.Payload, memberKey) + member, _ = Approve(member, root.Payload.PeerID, rootKey) + if err := Save(repositoryPath, member); err != nil { + t.Fatal(err) + } + beforeDelete := strings.TrimSpace(gitMembershipRunOutput(t, repositoryPath, "rev-parse", SharedRef)) + if err := removeSharedFile(repositoryPath, membersPrefix+member.Payload.PeerID+".json"); err != nil { + t.Fatal(err) + } + afterDelete := strings.TrimSpace(gitMembershipRunOutput(t, repositoryPath, "rev-parse", SharedRef)) + if err := ValidateControlUpdates(repositoryPath, map[string]RefUpdate{SharedRef: {Old: beforeDelete, New: afterDelete}}); err == nil { + t.Fatal("managed receive accepted membership document deletion") + } +} + +func TestRevokedApproverCannotKeepAdmissionChainAlive(t *testing.T) { + repositoryPath := gitMembershipRepository(t) + filesystemID := strings.Repeat("c", 40) + rootKey, rootPublic, _ := EnsureKey(t.TempDir()) + adminKey, adminPublic, _ := EnsureKey(t.TempDir()) + memberKey, memberPublic, _ := EnsureKey(t.TempDir()) + root := signedTestRecord(t, filesystemID, "root-peer", "admin", rootPublic, rootKey) + root, _ = Approve(root, "root-peer", rootKey) + admin := signedTestRecord(t, filesystemID, "other-admin", "admin", adminPublic, adminKey) + admin, _ = Approve(admin, "root-peer", rootKey) + member := signedTestRecord(t, filesystemID, "member-peer", "member", memberPublic, memberKey) + member, _ = Approve(member, "other-admin", adminKey) + for _, record := range []Record{root, admin, member} { + if err := Save(repositoryPath, record); err != nil { + t.Fatal(err) + } + } + if err := Trust(repositoryPath, root.Payload.PeerID, rootPublic); err != nil { + t.Fatal(err) + } + revocation, _ := Revoke(filesystemID, admin.Payload.PeerID, root.Payload.PeerID, 2, rootKey) + if err := SaveRevocation(repositoryPath, revocation); err != nil { + t.Fatal(err) + } + unauthorized, _ := Revoke(filesystemID, member.Payload.PeerID, admin.Payload.PeerID, 2, adminKey) + if err := SaveRevocation(repositoryPath, unauthorized); err != nil { + t.Fatal(err) + } + accepted, err := Accepted(repositoryPath, filesystemID) + if err != nil { + t.Fatal(err) + } + if len(accepted) != 1 || accepted[0].Payload.PeerID != "root-peer" { + t.Fatalf("revoked approver chain remained accepted: %#v", accepted) + } + revoked, err := AcceptedRevocations(repositoryPath, filesystemID) + if err != nil { + t.Fatal(err) + } + if revoked[member.Payload.PeerID] { + t.Fatal("revoked administrator issued a new sticky revocation") + } +} + func TestCurrentMachinesSelectsNewestReinstalledPeer(t *testing.T) { base := time.Date(2026, 8, 19, 6, 0, 0, 0, time.UTC) records := []Record{ diff --git a/internal/mount/filesystem.go b/internal/mount/filesystem.go index 2324b76..fc9023d 100644 --- a/internal/mount/filesystem.go +++ b/internal/mount/filesystem.go @@ -38,6 +38,7 @@ type FileSystem struct { logger *slog.Logger writesMu sync.Mutex writes map[string]*writeSession + starting map[string]chan struct{} closing map[string]chan struct{} cacheInvalidator contentInvalidator } @@ -87,7 +88,7 @@ func NewFileSystemWithContext(ctx context.Context, api core.API, notifier change logger = slog.New(slog.NewTextHandler(io.Discard, nil)) } return &FileSystem{FileSystem: pathfs.NewDefaultFileSystem(), core: api, lifetime: ctx, - notifier: notifier, logger: logger, writes: make(map[string]*writeSession), closing: make(map[string]chan struct{})} + notifier: notifier, logger: logger, writes: make(map[string]*writeSession), starting: make(map[string]chan struct{}), closing: make(map[string]chan struct{})} } func status(err error) fuse.Status { @@ -362,6 +363,9 @@ func (f *FileSystem) openWrite(path string, flags uint32, create bool, modes ... for { f.writesMu.Lock() done := f.closing[path] + if done == nil { + done = f.starting[path] + } if done == nil { break } @@ -390,6 +394,12 @@ func (f *FileSystem) openWrite(path string, flags uint32, create bool, modes ... session.mu.Unlock() return &adapterFile{File: nodefs.NewDefaultFile(), filesystem: f, session: session, writable: writable}, fuse.OK } + started := make(chan struct{}) + f.starting[path] = started + f.writesMu.Unlock() + if f.notifier != nil { + f.notifier.BeginWrite() + } mode := uint32(0o644) if len(modes) > 0 { mode = modes[0] @@ -397,7 +407,13 @@ func (f *FileSystem) openWrite(path string, flags uint32, create bool, modes ... request := core.WriteRequest{Path: path, Mode: mode, Create: create, Exclusive: create && flags&syscall.O_EXCL != 0, Truncate: flags&syscall.O_TRUNC != 0} transaction, err := f.core.BeginWrite(f.lifetime, request) if err != nil { + f.writesMu.Lock() + delete(f.starting, path) + close(started) f.writesMu.Unlock() + if f.notifier != nil { + f.notifier.EndWrite() + } return nil, status(err) } attributes := core.Attributes{Kind: core.KindFile, Mode: syscall.S_IFREG | mode&0o7777, Modified: time.Now(), Changed: time.Now(), Accessed: time.Now()} @@ -410,11 +426,11 @@ func (f *FileSystem) openWrite(path string, flags uint32, create bool, modes ... attributes.Size = 0 } session := &writeSession{transaction: transaction, path: path, attributes: attributes, refs: 1, dirty: create || request.Truncate, created: create} + f.writesMu.Lock() f.writes[path] = session + delete(f.starting, path) + close(started) f.writesMu.Unlock() - if f.notifier != nil { - f.notifier.BeginWrite() - } return &adapterFile{File: nodefs.NewDefaultFile(), filesystem: f, session: session, writable: true}, fuse.OK } @@ -656,27 +672,32 @@ func (a *adapterFile) Allocate(offset, size uint64, mode uint32) fuse.Status { func (a *adapterFile) GetAttr(out *fuse.Attr) fuse.Status { a.session.mu.Lock() defer a.session.mu.Unlock() + if a.session.failure != nil { + return status(a.session.failure) + } *out = *fuseAttr(a.session.attributes) return fuse.OK } func (a *adapterFile) Flush() fuse.Status { - a.session.mu.Lock() - defer a.session.mu.Unlock() - return status(a.session.failure) + return a.commitDirty("flush") } func (a *adapterFile) Fsync(int) fuse.Status { + return a.commitDirty("fsync") +} + +func (a *adapterFile) commitDirty(reason string) fuse.Status { a.session.mu.Lock() defer a.session.mu.Unlock() - if a.session.failure != nil || !a.session.dirty { + if a.session.failure != nil || !a.session.dirty || a.session.removed { return status(a.session.failure) } if err := a.session.transaction.Commit(a.filesystem.lifetime); err != nil { a.session.failure = err return status(err) } - a.filesystem.changed("fsync", "path", a.session.path) + a.filesystem.changed(reason, "path", a.session.path) replacement, err := a.filesystem.core.BeginWrite(a.filesystem.lifetime, core.WriteRequest{Path: a.session.path}) if err != nil { a.session.failure = err @@ -710,12 +731,18 @@ func (f *FileSystem) releaseWrite(session *writeSession) { session.mu.Lock() var err error - if session.removed || session.failure != nil || !session.dirty { + if session.failure != nil { + err = session.failure + _ = session.transaction.Abort(context.Background()) + } else if session.removed || !session.dirty { err = session.transaction.Abort(context.Background()) } else { commitCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) err = session.transaction.Commit(commitCtx) cancel() + if err != nil { + session.failure = err + } } changed := err == nil && session.dirty && !session.removed session.mu.Unlock() diff --git a/internal/mount/filesystem_test.go b/internal/mount/filesystem_test.go index 4e6c607..189d0de 100644 --- a/internal/mount/filesystem_test.go +++ b/internal/mount/filesystem_test.go @@ -34,6 +34,22 @@ type blockingCommitTransaction struct { finish chan struct{} } +type blockingBeginAPI struct { + core.API + started chan struct{} + release chan struct{} +} + +func (a blockingBeginAPI) BeginWrite(ctx context.Context, request core.WriteRequest) (core.WriteTransaction, error) { + close(a.started) + select { + case <-a.release: + return a.API.BeginWrite(ctx, request) + case <-ctx.Done(): + return nil, ctx.Err() + } +} + func (t *blockingCommitTransaction) Commit(ctx context.Context) error { close(t.started) select { @@ -133,6 +149,36 @@ func TestCreateIsVisibleBeforeCloseAndPublishedAtomically(t *testing.T) { } } +func TestFlushReportsPublishFailureAndQuarantinesPayload(t *testing.T) { + root := t.TempDir() + filesystem, _, _ := testFileSystem(t, root) + if err := os.Mkdir(filepath.Join(root, "parent"), 0o755); err != nil { + t.Fatal(err) + } + handle, code := filesystem.Create("parent/item.txt", syscall.O_WRONLY, 0o644, nil) + if code != fuse.OK { + t.Fatal(code) + } + if _, code := handle.Write([]byte("recover me"), 0); code != fuse.OK { + t.Fatal(code) + } + if err := os.Remove(filepath.Join(root, "parent")); err != nil { + t.Fatal(err) + } + if code := handle.Flush(); code == fuse.OK { + t.Fatal("flush hid the failed final publication") + } + 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) + } + handle.Release() +} + func TestMultipleHandlesShareOneWriteUntilFinalRelease(t *testing.T) { root := t.TempDir() filesystem, _, _ := testFileSystem(t, root) @@ -193,6 +239,45 @@ func TestSlowWriteCommitDoesNotBlockUnrelatedDirectoryReads(t *testing.T) { } } +func TestSlowBeginWriteDoesNotHoldGlobalSessionLock(t *testing.T) { + root := t.TempDir() + filesystem, _, service := testFileSystem(t, root) + if err := os.WriteFile(filepath.Join(root, "existing.txt"), []byte("content"), 0o644); err != nil { + t.Fatal(err) + } + started := make(chan struct{}) + release := make(chan struct{}) + filesystem.core = blockingBeginAPI{API: service, started: started, release: release} + opened := make(chan fuse.Status, 1) + go func() { + _, status := filesystem.Open("existing.txt", syscall.O_RDWR, nil) + opened <- status + }() + <-started + readDone := make(chan fuse.Status, 1) + go func() { + _, status := filesystem.GetAttr("other.txt", nil) + readDone <- status + }() + select { + case status := <-readDone: + if status != fuse.ENOENT { + t.Fatalf("unrelated lookup status = %v", status) + } + case <-time.After(time.Second): + t.Fatal("slow BeginWrite held the global write-session lock") + } + close(release) + select { + case status := <-opened: + if status != fuse.OK { + t.Fatal(status) + } + case <-time.After(time.Second): + t.Fatal("write open did not finish") + } +} + func TestBusyWriteSessionDoesNotBlockDirectoryReads(t *testing.T) { filesystem, _, _ := testFileSystem(t, t.TempDir()) session := &writeSession{path: "busy.txt", refs: 1, dirty: true} diff --git a/internal/mount/host.go b/internal/mount/host.go index 6bb0ec7..7d67c71 100644 --- a/internal/mount/host.go +++ b/internal/mount/host.go @@ -37,8 +37,8 @@ type Options struct { type daemonNotifier struct{ repository string } func (n daemonNotifier) Notify(reason string) { _ = wakeup.Notify(n.repository, reason) } -func (daemonNotifier) BeginWrite() {} -func (daemonNotifier) EndWrite() {} +func (n daemonNotifier) BeginWrite() { _ = wakeup.Notify(n.repository, wakeup.WriterBegin) } +func (n daemonNotifier) EndWrite() { _ = wakeup.Notify(n.repository, wakeup.WriterEnd) } type fuseLogWriter struct{ logger *slog.Logger } diff --git a/internal/mount/host_test.go b/internal/mount/host_test.go index 8f72715..2492c37 100644 --- a/internal/mount/host_test.go +++ b/internal/mount/host_test.go @@ -11,9 +11,39 @@ import ( "testing" "time" + "github.com/bitbeamer/dfs/internal/wakeup" "github.com/hanwen/go-fuse/v2/fuse" ) +func TestDaemonNotifierCarriesWriterLifecycle(t *testing.T) { + repositoryPath := t.TempDir() + listener, err := wakeup.Listen(repositoryPath) + if err != nil { + t.Fatal(err) + } + defer listener.Close() + notifier := daemonNotifier{repository: repositoryPath} + messages := make(chan string, 2) + go func() { + for range 2 { + message, _ := listener.Receive() + messages <- message + } + }() + notifier.BeginWrite() + notifier.EndWrite() + for _, want := range []string{wakeup.WriterBegin, wakeup.WriterEnd} { + select { + case got := <-messages: + if got != want { + t.Fatalf("writer lifecycle message = %q, want %q", got, want) + } + case <-time.After(time.Second): + t.Fatalf("writer lifecycle message %q was not delivered", want) + } + } +} + type entryNotifyCall struct { directory string name string diff --git a/internal/mount/recovery.go b/internal/mount/recovery.go index 5c75734..66dc5b9 100644 --- a/internal/mount/recovery.go +++ b/internal/mount/recovery.go @@ -149,31 +149,27 @@ func recoverStartup(ctx context.Context, repo *repository.Repository, mountpoint } run.logger.Warn("abrupt prior mount detected", "recovery_directory", run.batch) } - if err := run.recoverWrites(); err != nil { - return fail(err) - } - if err := run.quarantineGitLocks(); err != nil { - return fail(err) - } - if err := run.quarantineAnnexTemps(); err != nil { - return fail(err) - } - if err := repo.RepairLegacyPrivateState(ctx); err != nil { - return fail(fmt.Errorf("remove legacy DFS state from Git: %w", err)) - } - operation, err := run.snapshotIncompleteGitOperation() - if err != nil { + if err := repo.RecoverWorkTree(ctx, func() error { + if err := run.recoverWrites(); err != nil { + return err + } + if err := run.quarantineGitLocks(); err != nil { + return err + } + if err := run.quarantineAnnexTemps(); err != nil { + return err + } + operation, err := run.snapshotIncompleteGitOperation() + if err != nil { + return err + } + if operation != "" { + return fmt.Errorf("incomplete Git operation %s was preserved in %s; resolve or abort it before mounting", operation, run.batch) + } + return nil + }); err != nil { return fail(err) } - if operation != "" { - return fail(fmt.Errorf("incomplete Git operation %s was preserved in %s; resolve or abort it before mounting", operation, run.batch)) - } - if _, err := repo.CommitPending(ctx, "Recover interrupted DFS update"); err != nil { - return fail(fmt.Errorf("finish pending repository update: %w", err)) - } - if err := repo.CheckConsistency(ctx); err != nil { - return fail(fmt.Errorf("repository consistency check failed: %w", err)) - } if run.quarantined > 0 { run.logger.Warn("startup recovery completed", "quarantined", run.quarantined, "directory", run.batch) } else { diff --git a/internal/peer/client.go b/internal/peer/client.go index 9e8e22c..1d9c13e 100644 --- a/internal/peer/client.go +++ b/internal/peer/client.go @@ -164,7 +164,8 @@ func PairAndJoinWithOptions(ctx context.Context, encodedInvitation, destination, if start.Approver.Payload.PeerID != start.PeerID || start.Membership.Payload.PeerID != peerID { return nil, errors.New("pairing peer returned mismatched membership identities") } - if err := verifyMembershipApprovalChain(start.Approver, start.Members, invitation.FileSystemID); err != nil { + bootstrapAdmin, err := membershipApprovalRoot(start.Approver, start.Members, invitation.FileSystemID) + if err != nil { return nil, fmt.Errorf("verify approving DFS membership: %w", err) } if err := membership.VerifyApproval(start.Membership, start.Approver); err != nil { @@ -211,6 +212,9 @@ func PairAndJoinWithOptions(ctx context.Context, encodedInvitation, destination, if err := membership.Save(repo.Config.Repository, start.Approver); err != nil { return nil, fmt.Errorf("save approving DFS membership: %w", err) } + if err := membership.TrustBootstrapAdmin(repo.Config.Repository, bootstrapAdmin); err != nil { + return nil, fmt.Errorf("trust bootstrap DFS administrator: %w", err) + } if err := membership.Save(repo.Config.Repository, start.Membership); err != nil { return nil, fmt.Errorf("save local DFS membership: %w", err) } @@ -286,13 +290,18 @@ func PairAndJoinWithOptions(ctx context.Context, encodedInvitation, destination, } func verifyMembershipApprovalChain(record membership.Record, members []membership.Record, filesystemID string) error { + _, err := membershipApprovalRoot(record, members, filesystemID) + return err +} + +func membershipApprovalRoot(record membership.Record, members []membership.Record, filesystemID string) (membership.Record, error) { byID := make(map[string]membership.Record, len(members)+1) for _, member := range append(append([]membership.Record(nil), members...), record) { if member.Payload.FileSystemID != filesystemID { continue } if existing, found := byID[member.Payload.PeerID]; found && existing.Payload.SigningPublicKey != member.Payload.SigningPublicKey { - return fmt.Errorf("membership chain contains conflicting records for peer %s", member.Payload.PeerID) + return membership.Record{}, fmt.Errorf("membership chain contains conflicting records for peer %s", member.Payload.PeerID) } byID[member.Payload.PeerID] = member } @@ -301,21 +310,24 @@ func verifyMembershipApprovalChain(record membership.Record, members []membershi for { peerID := current.Payload.PeerID if current.Payload.FileSystemID != filesystemID { - return errors.New("approving membership belongs to another filesystem") + return membership.Record{}, errors.New("approving membership belongs to another filesystem") } if visited[peerID] { - return errors.New("membership approval chain contains a cycle") + return membership.Record{}, errors.New("membership approval chain contains a cycle") } visited[peerID] = true approver, found := byID[current.ApprovedBy] if !found { - return fmt.Errorf("membership approval chain is missing peer %s", current.ApprovedBy) + return membership.Record{}, fmt.Errorf("membership approval chain is missing peer %s", current.ApprovedBy) } if err := membership.VerifyApproval(current, approver); err != nil { - return err + return membership.Record{}, err + } + if approver.Payload.Role != "admin" { + return membership.Record{}, fmt.Errorf("membership approver %s is not an administrator", approver.Payload.PeerID) } if current.ApprovedBy == peerID { - return nil + return current, nil } current = approver } diff --git a/internal/peer/diagnostic.go b/internal/peer/diagnostic.go index 5b7ea46..77734e6 100644 --- a/internal/peer/diagnostic.go +++ b/internal/peer/diagnostic.go @@ -289,11 +289,7 @@ func CheckMesh(ctx context.Context, repo *repository.Repository, discoveryTimeou } } if offers, discoverErr := Discover(ctx, discoveryTimeout); discoverErr == nil { - for _, offer := range offers { - if offer.FileSystemID == filesystemID && offer.PeerID != repo.Config.PeerID { - peers[offer.PeerID] = MeshPeer{PeerID: offer.PeerID, PeerName: offer.PeerName, Online: true} - } - } + markDiscoveredAcceptedPeers(peers, offers, filesystemID, repo.Config.PeerID) } remotes, err := repo.Remotes(ctx) if err != nil { @@ -358,6 +354,18 @@ func CheckMesh(ctx context.Context, repo *repository.Repository, discoveryTimeou return evaluateMesh(peers, reports, reportErrors), nil } +func markDiscoveredAcceptedPeers(peers map[string]MeshPeer, offers []Offer, filesystemID, localPeerID string) { + for _, offer := range offers { + if offer.FileSystemID != filesystemID || offer.PeerID == localPeerID { + continue + } + if member, accepted := peers[offer.PeerID]; accepted { + member.Online = true + peers[offer.PeerID] = member + } + } +} + func meshPeerIDForRemote(peers map[string]MeshPeer, remoteName string) string { for peerID := range peers { if pairedRemoteName(peerID) == remoteName { diff --git a/internal/peer/invitation.go b/internal/peer/invitation.go index 4d9d22b..a395ce3 100644 --- a/internal/peer/invitation.go +++ b/internal/peer/invitation.go @@ -63,6 +63,20 @@ func createInvitation(repo *repository.Repository, lifetime time.Duration, bound if err != nil { return Invitation{}, err } + accepted, err := membership.Accepted(repo.Config.Repository, filesystemID, repo.Config.PeerID) + if err != nil { + return Invitation{}, err + } + administrator := false + for _, record := range accepted { + if record.Payload.PeerID == repo.Config.PeerID && record.Payload.Role == "admin" { + administrator = true + break + } + } + if !administrator { + return Invitation{}, errors.New("only an administrator member can create pairing invitations") + } _, fingerprint, err := loadOrCreateCertificate(repo.Config.Repository) if err != nil { return Invitation{}, err diff --git a/internal/peer/membership.go b/internal/peer/membership.go index 8a80eb3..3b4025e 100644 --- a/internal/peer/membership.go +++ b/internal/peer/membership.go @@ -53,7 +53,7 @@ func ensureLocalMembership(ctx context.Context, repo *repository.Repository, fil if err := membership.Save(repo.Config.Repository, record); err != nil { return nil, membership.Record{}, err } - if err := membership.Trust(repo.Config.Repository, record.Payload.PeerID, record.Payload.SigningPublicKey); err != nil { + if err := membership.TrustBootstrapAdmin(repo.Config.Repository, record); err != nil { return nil, membership.Record{}, err } return private, record, nil @@ -88,6 +88,9 @@ func validatePairingMembership(record membership.Record, filesystemID, peerID, n if payload.FileSystemID != filesystemID || payload.PeerID != peerID || payload.Name != strings.TrimSpace(name) { return errors.New("pairing membership does not match the authenticated peer") } + if payload.Role != "member" { + return errors.New("new peers must request the member role") + } if !strings.HasPrefix(payload.QUICEndpoint, "quic://") { return errors.New("pairing membership has an invalid QUIC endpoint") } @@ -194,11 +197,8 @@ func RevokeMembership(ctx context.Context, repo *repository.Repository, remote s if err != nil { return err } - var local, target membership.Record + var target membership.Record for _, record := range records { - if record.Payload.PeerID == repo.Config.PeerID { - local = record - } if strings.HasPrefix(record.Payload.PeerID, peerPrefix) { target = record } @@ -206,6 +206,17 @@ func RevokeMembership(ctx context.Context, repo *repository.Repository, remote s if target.Payload.PeerID == "" { return fmt.Errorf("membership for %s was not found", remote) } + accepted, err := membership.Accepted(repo.Config.Repository, filesystemID, repo.Config.PeerID) + if err != nil { + return err + } + var local membership.Record + for _, record := range accepted { + if record.Payload.PeerID == repo.Config.PeerID { + local = record + break + } + } if local.Payload.Role != "admin" { return errors.New("only an administrator member can revoke a DFS peer") } diff --git a/internal/peer/peer_test.go b/internal/peer/peer_test.go index 3f0b5f8..fecf678 100644 --- a/internal/peer/peer_test.go +++ b/internal/peer/peer_test.go @@ -96,6 +96,39 @@ func TestReachableOffersDropsStaleAdvertisements(t *testing.T) { } } +func TestMeshDiscoveryCannotIntroduceUnacceptedPeer(t *testing.T) { + filesystemID := strings.Repeat("a", 40) + peers := map[string]MeshPeer{ + "accepted-peer": {PeerID: "accepted-peer", PeerName: "accepted"}, + } + markDiscoveredAcceptedPeers(peers, []Offer{ + {FileSystemID: filesystemID, PeerID: "accepted-peer"}, + {FileSystemID: filesystemID, PeerID: "unknown-peer"}, + }, filesystemID, "local-peer") + if !peers["accepted-peer"].Online { + t.Fatal("accepted discovered peer was not marked online") + } + if _, found := peers["unknown-peer"]; found { + t.Fatal("unauthenticated discovery introduced an unknown mesh peer") + } +} + +func TestValidatePairingMembershipRejectsAdminClaim(t *testing.T) { + filesystemID := strings.Repeat("a", 40) + private, draft, err := newMembershipDraft(filepath.Join(t.TempDir(), "draft"), filesystemID, "joining-peer", "joining", 7843) + if err != nil { + t.Fatal(err) + } + draft.Payload.Role = "admin" + draft, err = membership.Sign(draft.Payload, private) + if err != nil { + t.Fatal(err) + } + if err := validatePairingMembership(draft, filesystemID, "joining-peer", "joining"); err == nil { + t.Fatal("pairing accepted an administrator role claim") + } +} + func TestMDNSHostname(t *testing.T) { for input, expected := range map[string]string{ "cachyos": "cachyos.local.", @@ -255,6 +288,11 @@ func TestPairAndJoinConfiguresBothPeers(t *testing.T) { if joinedID != existingID || result.Repository.Config.PeerID == existing.Config.PeerID { t.Fatalf("paired identities existing=%q joined=%q peers=%q/%q", existingID, joinedID, existing.Config.PeerID, result.Repository.Config.PeerID) } + if _, record, err := ensureLocalMembership(ctx, result.Repository, joinedID, DefaultPairingPort); err != nil { + t.Fatalf("restart paired member service: %v", err) + } else if record.Payload.Role != "member" { + t.Fatalf("restarted paired role = %q, want member", record.Payload.Role) + } remotes, err := existing.Remotes(ctx) if err != nil { t.Fatal(err) @@ -461,11 +499,21 @@ func TestVerifyMembershipApprovalChainAcceptsNonFounderApprover(t *testing.T) { if err != nil { t.Fatal(err) } + root.Payload.Role = "admin" + root, err = membership.Sign(root.Payload, rootKey) + if err != nil { + t.Fatal(err) + } root, err = membership.Approve(root, root.Payload.PeerID, rootKey) if err != nil { t.Fatal(err) } - _, approver, err := newMembershipDraft(filepath.Join(t.TempDir(), "approver"), filesystemID, "approver-peer", "approver", 7844) + approverKey, approver, err := newMembershipDraft(filepath.Join(t.TempDir(), "approver"), filesystemID, "approver-peer", "approver", 7844) + if err != nil { + t.Fatal(err) + } + approver.Payload.Role = "admin" + approver, err = membership.Sign(approver.Payload, approverKey) if err != nil { t.Fatal(err) } @@ -476,6 +524,9 @@ func TestVerifyMembershipApprovalChainAcceptsNonFounderApprover(t *testing.T) { if err := verifyMembershipApprovalChain(approver, []membership.Record{root, approver}, filesystemID); err != nil { t.Fatalf("valid non-founder approval chain rejected: %v", err) } + if bootstrap, err := membershipApprovalRoot(approver, []membership.Record{root, approver}, filesystemID); err != nil || bootstrap.Payload.PeerID != root.Payload.PeerID { + t.Fatalf("approval root = %#v, %v", bootstrap, err) + } if err := verifyMembershipApprovalChain(approver, []membership.Record{approver}, filesystemID); err == nil || !strings.Contains(err.Error(), "missing peer") { t.Fatalf("incomplete approval chain error = %v", err) } diff --git a/internal/peer/service.go b/internal/peer/service.go index 2ce8d46..7b80e08 100644 --- a/internal/peer/service.go +++ b/internal/peer/service.go @@ -319,6 +319,10 @@ func (s *Service) handlePairStart(response http.ResponseWriter, request *http.Re writeProtocolError(response, http.StatusInternalServerError, "cannot create pairing session") return } + if s.membershipRecord.Payload.Role != "admin" { + writeProtocolError(response, http.StatusForbidden, "pairing approval requires an administrator") + return + } approvedMembership, err := membership.Approve(input.Membership, s.repo.Config.PeerID, s.membershipKey) if err != nil { writeProtocolError(response, http.StatusInternalServerError, "cannot approve peer membership") diff --git a/internal/repository/convergence_integration_test.go b/internal/repository/convergence_integration_test.go index 36de490..6ecb6b6 100644 --- a/internal/repository/convergence_integration_test.go +++ b/internal/repository/convergence_integration_test.go @@ -5,6 +5,7 @@ import ( "os" "os/exec" "path/filepath" + "slices" "sort" "strings" "testing" @@ -144,6 +145,65 @@ func TestTwoPeerConcurrentChangesConverge(t *testing.T) { }) } +func TestApplyReceivedRetainsBothConflictingInboxEdits(t *testing.T) { + if _, err := exec.LookPath("git-annex"); err != nil { + t.Skip("git-annex is not installed") + } + linux, mac, ctx := newConvergencePeers(t, "inbox.txt", "baseline\n") + writePeerFile(t, ctx, linux, "inbox.txt", "linux edit\n") + writePeerFile(t, ctx, mac, "inbox.txt", "mac edit\n") + commitPeerChange(t, ctx, linux) + commitPeerChange(t, ctx, mac) + wantKeys := []string{ + gitOutput(t, ctx, linux, "show", "HEAD:inbox.txt"), + gitOutput(t, ctx, mac, "show", "HEAD:inbox.txt"), + } + inbox := "refs/heads/dfs-incoming/" + linux.Config.PeerID + "/main" + if _, err := mac.runner.Run(ctx, "git", "fetch", "--no-tags", linux.Config.Repository, "main:"+inbox); err != nil { + t.Fatal(err) + } + if err := mac.ApplyReceived(ctx); err != nil { + t.Fatal(err) + } + paths := pathsForConflict(trackedPaths(t, ctx, mac), "inbox.txt") + if len(paths) != 2 { + t.Fatalf("inbox conflict paths = %q, want original and one variant", paths) + } + gotKeys := []string{ + gitOutput(t, ctx, mac, "show", "HEAD:"+paths[0]), + gitOutput(t, ctx, mac, "show", "HEAD:"+paths[1]), + } + sort.Strings(gotKeys) + sort.Strings(wantKeys) + if !slices.Equal(gotKeys, wantKeys) { + t.Fatalf("inbox conflict keys = %q, want %q", gotKeys, wantKeys) + } +} + +func TestCommitPendingRefusesUnresolvedMerge(t *testing.T) { + if _, err := exec.LookPath("git-annex"); err != nil { + t.Skip("git-annex is not installed") + } + first, second, ctx := newConvergencePeers(t, "conflict.txt", "baseline\n") + writePeerFile(t, ctx, first, "conflict.txt", "first edit\n") + writePeerFile(t, ctx, second, "conflict.txt", "second edit\n") + commitPeerChange(t, ctx, first) + commitPeerChange(t, ctx, second) + other := "refs/heads/test-conflicting-peer" + if _, err := second.runner.Run(ctx, "git", "fetch", "--no-tags", first.Config.Repository, "main:"+other); err != nil { + t.Fatal(err) + } + if _, err := second.runner.Run(ctx, "git", "merge", "--no-edit", other); err == nil { + t.Fatal("conflicting merge unexpectedly succeeded") + } + if _, err := second.CommitPending(ctx, "must not resolve conflict"); err == nil { + t.Fatal("CommitPending accepted an unresolved merge") + } + if _, err := os.Stat(filepath.Join(second.Config.Repository, ".git", "MERGE_HEAD")); err != nil { + t.Fatalf("CommitPending destroyed merge evidence: %v", err) + } +} + func newConvergencePeers(t *testing.T, path, content string) (*Repository, *Repository, context.Context) { t.Helper() root := t.TempDir() @@ -151,7 +211,8 @@ func newConvergencePeers(t *testing.T, path, content string) (*Repository, *Repo ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) t.Cleanup(cancel) - linux, err := Init(ctx, filepath.Join(root, "linux"), "linux", 10<<20) + identity := GitIdentity{Name: "DFS Convergence Test", Email: "convergence@example.invalid"} + linux, err := InitWithIdentity(ctx, filepath.Join(root, "linux"), "linux", 10<<20, identity) if err != nil { t.Fatal(err) } @@ -159,7 +220,7 @@ func newConvergencePeers(t *testing.T, path, content string) (*Repository, *Repo writePeerFile(t, ctx, linux, path, content) commitPeerChange(t, ctx, linux) - mac, err := Join(ctx, linux.Config.Repository, filepath.Join(root, "mac"), "mac", 10<<20) + mac, err := JoinWithIdentity(ctx, linux.Config.Repository, filepath.Join(root, "mac"), "mac", 10<<20, identity) if err != nil { t.Fatal(err) } diff --git a/internal/repository/repository.go b/internal/repository/repository.go index c5eedd6..c20059c 100644 --- a/internal/repository/repository.go +++ b/internal/repository/repository.go @@ -79,8 +79,8 @@ func (e *RemoteSyncError) Error() string { } type Remote struct { - Name string - URL string + Name string `json:"name"` + URL string `json:"url"` } type StorageRemote struct { @@ -483,6 +483,41 @@ func (r *Repository) WithWorkTreeLock(fn func() error) error { return fn() } +// BeginWriteGuard prevents received commits from replacing the worktree while +// a frontend write transaction is open. Multiple local writers may coexist; +// ApplyReceived takes the corresponding exclusive lock. +func (r *Repository) BeginWriteGuard(ctx context.Context) (func(), error) { + return r.lockWriterProcess(ctx, unix.LOCK_SH) +} + +func (r *Repository) lockWriterProcess(ctx context.Context, mode int) (func(), error) { + path := filepath.Join(r.Config.Repository, filepath.FromSlash(config.Directory), "writers.lock") + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + return nil, err + } + file, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o600) + if err != nil { + return nil, err + } + for { + if err := unix.Flock(int(file.Fd()), mode|unix.LOCK_NB); err == nil { + return func() { + _ = unix.Flock(int(file.Fd()), unix.LOCK_UN) + _ = file.Close() + }, nil + } else if !errors.Is(err, unix.EWOULDBLOCK) && !errors.Is(err, unix.EAGAIN) { + _ = file.Close() + return nil, err + } + select { + case <-ctx.Done(): + _ = file.Close() + return nil, ctx.Err() + case <-time.After(25 * time.Millisecond): + } + } +} + func (r *Repository) CommitPending(ctx context.Context, message string) (bool, error) { r.mu.Lock() defer r.mu.Unlock() @@ -545,10 +580,51 @@ func (r *Repository) CheckConsistency(ctx context.Context) error { return nil } +// RecoverWorkTree runs startup recovery and the follow-up Git repair as one +// cross-process critical section, so recovery never quarantines locks or annex +// state belonging to a live daemon operation. +func (r *Repository) RecoverWorkTree(ctx context.Context, prepare func() error) error { + r.mu.Lock() + defer r.mu.Unlock() + unlock, err := r.lockWorkTreeProcess(ctx) + if err != nil { + return err + } + defer unlock() + if err := prepare(); err != nil { + return err + } + if err := r.repairLegacyPrivateStateLocked(ctx); err != nil { + return fmt.Errorf("remove legacy DFS state from Git: %w", err) + } + if _, err := r.commitPendingLocked(ctx, "Recover interrupted DFS update"); err != nil { + return fmt.Errorf("finish pending repository update: %w", err) + } + if _, err := r.runner.Run(ctx, "git", "fsck", "--no-dangling"); err != nil { + return err + } + if _, err := r.runner.Run(ctx, "git", "annex", "fsck", "--fast"); err != nil { + return err + } + return nil +} + func (r *Repository) commitPendingLocked(ctx context.Context, message string) (bool, error) { if err := r.repairLegacyPrivateStateLocked(ctx); err != nil { return false, err } + unmerged, err := r.runner.Run(ctx, "git", "diff", "--name-only", "--diff-filter=U") + if err != nil { + return false, err + } + if strings.TrimSpace(unmerged) != "" { + return false, fmt.Errorf("refuse to commit with unresolved merge conflicts: %s", strings.Join(strings.Fields(unmerged), ", ")) + } + if _, err := os.Stat(filepath.Join(r.Config.Repository, ".git", "MERGE_HEAD")); err == nil { + return false, errors.New("refuse to commit while a merge is in progress") + } else if !errors.Is(err, os.ErrNotExist) { + return false, err + } // git-annex handles new and modified user files. Git then records deletions, // renames, pointer updates, and ordinary control files. // git-annex normally skips dotfiles and files below dot-directories. The @@ -641,6 +717,11 @@ func (r *Repository) Sync(ctx context.Context, metadataOnly bool) error { // worktree. It deliberately performs no network I/O: the peer's receive-pack // has completed, and unrelated or offline peers must not delay visibility. func (r *Repository) ApplyReceived(ctx context.Context) error { + writersDrained, err := r.lockWriterProcess(ctx, unix.LOCK_EX) + if err != nil { + return err + } + defer writersDrained() r.mu.Lock() defer r.mu.Unlock() unlock, err := r.lockWorkTreeProcess(ctx) @@ -653,8 +734,22 @@ func (r *Repository) ApplyReceived(ctx context.Context) error { return err } for _, ref := range strings.Fields(refs) { - if _, err := r.runner.Run(ctx, "git", "-c", "merge.renames=false", "merge", "--no-edit", ref); err != nil { - return err + if _, mergeErr := r.runner.Run(ctx, "git", "-c", "merge.renames=false", "merge", "--no-edit", ref); mergeErr != nil { + // Resolve the active merge before any later commit path can observe it. + // git-annex retains concurrent file edits under distinct filenames. + if _, resolveErr := r.runner.Run(ctx, "git", "annex", "resolvemerge"); resolveErr != nil { + if _, abortErr := r.runner.Run(ctx, "git", "merge", "--abort"); abortErr != nil { + return errors.Join(mergeErr, resolveErr, fmt.Errorf("abort conflicting inbox merge: %w", abortErr)) + } + return errors.Join(mergeErr, resolveErr) + } + if _, statErr := os.Stat(filepath.Join(r.Config.Repository, ".git", "MERGE_HEAD")); statErr == nil { + if _, commitErr := r.runner.Run(ctx, "git", "commit", "--no-edit"); commitErr != nil { + return fmt.Errorf("commit resolved inbox merge: %w", commitErr) + } + } else if !os.IsNotExist(statErr) { + return statErr + } } } _, err = r.runner.Run(ctx, "git", "-c", "merge.renames=false", "annex", "sync", "--no-content", "--no-pull", "--no-push") diff --git a/internal/repository/repository_lock_test.go b/internal/repository/repository_lock_test.go index 9c32468..9d88fb7 100644 --- a/internal/repository/repository_lock_test.go +++ b/internal/repository/repository_lock_test.go @@ -1,8 +1,13 @@ package repository import ( + "context" + "errors" "testing" "time" + + "github.com/bitbeamer/dfs/internal/config" + "golang.org/x/sys/unix" ) func TestWithWorkTreeLockSerializesAccess(t *testing.T) { @@ -38,3 +43,62 @@ func TestWithWorkTreeLockSerializesAccess(t *testing.T) { t.Fatal("second worktree operation did not proceed after lock release") } } + +func TestRecoverWorkTreeWaitsForCrossProcessLockBeforeRecovery(t *testing.T) { + root := t.TempDir() + first := &Repository{Config: config.Default("first", root)} + second := &Repository{Config: config.Default("second", root)} + entered := make(chan struct{}) + release := make(chan struct{}) + go func() { + _ = first.WithWorkTreeLock(func() error { + close(entered) + <-release + return nil + }) + }() + <-entered + ctx, cancel := context.WithTimeout(context.Background(), 75*time.Millisecond) + defer cancel() + prepared := false + err := second.RecoverWorkTree(ctx, func() error { + prepared = true + return nil + }) + close(release) + if !errors.Is(err, context.DeadlineExceeded) || prepared { + t.Fatalf("recovery under live worktree lock = prepared %t, error %v", prepared, err) + } +} + +func TestReceivedChangesWaitForOpenWriterGuard(t *testing.T) { + root := t.TempDir() + writer := &Repository{Config: config.Default("writer", root)} + receiver := &Repository{Config: config.Default("receiver", root)} + unlock, err := writer.BeginWriteGuard(context.Background()) + if err != nil { + t.Fatal(err) + } + entered := make(chan struct{}) + done := make(chan struct{}) + go func() { + defer close(done) + release, lockErr := receiver.lockWriterProcess(context.Background(), unix.LOCK_EX) + if lockErr == nil { + close(entered) + release() + } + }() + select { + case <-entered: + t.Fatal("received-change guard entered while a writer was open") + case <-time.After(75 * time.Millisecond): + } + unlock() + select { + case <-entered: + case <-time.After(time.Second): + t.Fatal("received-change guard did not proceed after the writer closed") + } + <-done +} diff --git a/internal/repository/sync_plan.go b/internal/repository/sync_plan.go new file mode 100644 index 0000000..54c1d90 --- /dev/null +++ b/internal/repository/sync_plan.go @@ -0,0 +1,117 @@ +package repository + +import ( + "context" + "path/filepath" + "sort" + "strconv" + "strings" +) + +// SyncPlan describes what a synchronization pass would do, without changing +// anything. Eviction candidates assume inactive range-cache extents are +// reclaimed first and remain subject to git-annex copy-safety checks. +type SyncPlan struct { + Remotes []Remote `json:"remotes"` + Pins []PinnedPathHealth `json:"pins,omitempty"` + CacheBytes int64 `json:"cache_bytes"` + CacheLimit int64 `json:"cache_limit_bytes"` + Evictions []EvictionCandidate `json:"evictions,omitempty"` +} + +// EvictionCandidate is one unpinned cached file that cache enforcement would +// try to evict, least recently used first. +type EvictionCandidate struct { + Path string `json:"path"` + Size int64 `json:"size_bytes"` +} + +// SyncPlan computes the synchronization plan read-only: the remotes metadata +// would be exchanged with, the hydration state of every pinned path, and the +// LRU eviction candidates needed to bring the cache within its limit. +func (r *Repository) SyncPlan(ctx context.Context, metadataOnly bool) (*SyncPlan, error) { + remotes, err := r.Remotes(ctx) + if err != nil { + return nil, err + } + plan := &SyncPlan{Remotes: remotes} + if metadataOnly { + return plan, nil + } + pinRecords, err := r.Store.PinRecords() + if err != nil { + return nil, err + } + r.mu.Lock() + all, err := r.annexHealthFilesLocked(ctx) + if err != nil { + r.mu.Unlock() + return nil, err + } + local, err := r.annexHealthFilesHereLocked(ctx) + if err != nil { + r.mu.Unlock() + return nil, err + } + tree, err := r.runner.Run(ctx, "git", "ls-tree", "-r", "-z", "--format=%(objecttype)%x09%(objectsize)%x09%(path)", "HEAD") + if err != nil { + r.mu.Unlock() + return nil, err + } + usage, err := r.cacheUsageLocked() + r.mu.Unlock() + if err != nil { + return nil, err + } + annexed := make(map[string]annexHealthFile, len(all)) + fileSizes := make(map[string]int64, len(all)) + for _, file := range all { + annexed[file.Path] = file + fileSizes[file.Path] = file.Size + } + for _, entry := range strings.Split(tree, "\x00") { + if entry == "" { + continue + } + fields := strings.SplitN(entry, "\t", 3) + if len(fields) != 3 || fields[0] != "blob" { + continue + } + path := filepath.ToSlash(fields[2]) + if _, found := annexed[path]; found { + continue + } + if size, parseErr := strconv.ParseInt(fields[1], 10, 64); parseErr == nil { + fileSizes[path] = size + } + } + plan.Pins = pinnedPathHealth(pinRecords, fileSizes, local, annexed) + plan.CacheBytes = usage + plan.CacheLimit = r.Config.CacheLimit + if usage <= r.Config.CacheLimit { + return plan, nil + } + files, err := r.CachedFiles(ctx) + if err != nil { + return nil, err + } + sort.Slice(files, func(i, j int) bool { + return r.Store.LastAccess(files[i].Path).Before(r.Store.LastAccess(files[j].Path)) + }) + remaining := usage + for _, file := range files { + if remaining <= r.Config.CacheLimit { + break + } + pinned, err := r.Store.IsPinned(file.Path) + if err != nil { + return nil, err + } + if pinned { + continue + } + plan.Evictions = append(plan.Evictions, EvictionCandidate{Path: file.Path, Size: file.Size}) + remaining -= file.Size + } + return plan, nil +} diff --git a/internal/syncer/syncer.go b/internal/syncer/syncer.go index ba48486..bf2665b 100644 --- a/internal/syncer/syncer.go +++ b/internal/syncer/syncer.go @@ -21,19 +21,22 @@ const ( ) type Scheduler struct { - repo *repository.Repository - interval time.Duration - logger *slog.Logger - events chan string - stop chan struct{} - done chan struct{} - once sync.Once - writers atomic.Int64 - activeMu sync.Mutex - active string - cancel context.CancelFunc - entries EntryInvalidator - reconcile func(context.Context) error + repo *repository.Repository + interval time.Duration + logger *slog.Logger + events chan string + stop chan struct{} + done chan struct{} + once sync.Once + writers atomic.Int64 + activeMu sync.Mutex + active string + cancel context.CancelFunc + entries EntryInvalidator + reconcile func(context.Context) error + syncOverride func(string) + deferredMu sync.Mutex + deferredReason string } type EntryInvalidator interface { @@ -135,6 +138,15 @@ func (s *Scheduler) EndWrite() { writers = 0 } s.logger.Debug("writer closed", "open_writers", writers) + if writers == 0 { + s.deferredMu.Lock() + reason := s.deferredReason + s.deferredReason = "" + s.deferredMu.Unlock() + if reason != "" { + s.Notify(reason) + } + } } func (s *Scheduler) loop() { @@ -150,12 +162,22 @@ func (s *Scheduler) loop() { if debounce != nil { debounce.Stop() } - s.sync("shutdown") s.logger.Info("sync scheduler stopped") return case <-ticker.C: - if debounceC == nil { - s.sync("periodic") + if debounceC != nil { + if !debounce.Stop() { + select { + case <-debounce.C: + default: + } + } + debounceC = nil + pendingReason = mergeReasons(pendingReason, "periodic") + s.runSync(pendingReason) + pendingReason = "" + } else { + s.runSync("periodic") } case reason := <-s.events: if debounceC == nil { @@ -163,30 +185,38 @@ func (s *Scheduler) loop() { } else { pendingReason = mergeReasons(pendingReason, reason) } + if debounceC != nil { + continue + } if debounce == nil { debounce = time.NewTimer(eventDebounce) } else { - if !debounce.Stop() { - select { - case <-debounce.C: - default: - } - } debounce.Reset(eventDebounce) } debounceC = debounce.C case <-debounceC: debounceC = nil - s.sync(pendingReason) + s.runSync(pendingReason) pendingReason = "" } } } +func (s *Scheduler) runSync(reason string) { + if s.syncOverride != nil { + s.syncOverride(reason) + return + } + s.sync(reason) +} + func (s *Scheduler) sync(reason string) { writers := s.writers.Load() - if writers > 0 { + if writers > 0 && requiresWriterDrain(reason) { s.logger.Debug("automatic sync skipped", "reason", reason, "open_writers", writers) + s.deferredMu.Lock() + s.deferredReason = mergeReasons(s.deferredReason, reason) + s.deferredMu.Unlock() return } started := time.Now() @@ -305,6 +335,15 @@ func (s *Scheduler) sync(reason string) { } } +func requiresWriterDrain(reason string) bool { + switch reason { + case "managed Git receive", receiveAndPublishReason, maintenanceReceiveReason, "startup", "periodic", "shutdown": + return true + default: + return false + } +} + func isMaintenanceReason(reason string) bool { switch reason { case "startup", "periodic", "shutdown", "pin policy changed", "peer requested membership reconciliation", maintenanceReceiveReason: diff --git a/internal/syncer/syncer_test.go b/internal/syncer/syncer_test.go index 6aff1fc..d3c3843 100644 --- a/internal/syncer/syncer_test.go +++ b/internal/syncer/syncer_test.go @@ -6,8 +6,85 @@ import ( "io" "log/slog" "testing" + "time" ) +func TestEndWriteRequeuesDeferredReason(t *testing.T) { + scheduler := &Scheduler{ + logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + events: make(chan string, 1), + } + scheduler.BeginWrite() + scheduler.deferredReason = "completed write" + scheduler.EndWrite() + select { + case reason := <-scheduler.events: + if reason != "completed write" { + t.Fatalf("deferred reason = %q", reason) + } + case <-time.After(time.Second): + t.Fatal("deferred synchronization was not requeued") + } +} + +func TestOpenWriterOnlyDefersWorktreeChangingPasses(t *testing.T) { + for _, reason := range []string{"managed Git receive", "periodic", receiveAndPublishReason, maintenanceReceiveReason} { + if !requiresWriterDrain(reason) { + t.Errorf("%q does not wait for open writers", reason) + } + } + for _, reason := range []string{"completed write", "rename", "pin policy changed"} { + if requiresWriterDrain(reason) { + t.Errorf("push-only reason %q unnecessarily waits for open writers", reason) + } + } +} + +func TestContinuousEventsCannotStarveDebounce(t *testing.T) { + runs := make(chan string, 4) + scheduler := &Scheduler{ + interval: time.Hour, + logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + events: make(chan string, 128), + stop: make(chan struct{}), + done: make(chan struct{}), + syncOverride: func(reason string) { runs <- reason }, + } + go scheduler.loop() + deadline := time.Now().Add(700 * time.Millisecond) + for time.Now().Before(deadline) { + scheduler.Notify("filesystem change") + time.Sleep(50 * time.Millisecond) + } + select { + case <-runs: + case <-time.After(100 * time.Millisecond): + t.Fatal("continuous events postponed synchronization indefinitely") + } + close(scheduler.stop) + <-scheduler.done +} + +func TestShutdownDoesNotStartFreshSynchronization(t *testing.T) { + runs := make(chan string, 1) + scheduler := &Scheduler{ + interval: time.Hour, + logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + events: make(chan string, 1), + stop: make(chan struct{}), + done: make(chan struct{}), + syncOverride: func(reason string) { runs <- reason }, + } + go scheduler.loop() + close(scheduler.stop) + <-scheduler.done + select { + case reason := <-runs: + t.Fatalf("shutdown started a fresh %q synchronization", reason) + default: + } +} + func TestBeginWritePreemptsOutgoingSync(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) scheduler := &Scheduler{ diff --git a/internal/wakeup/wakeup.go b/internal/wakeup/wakeup.go index 4a7d41c..0b88e22 100644 --- a/internal/wakeup/wakeup.go +++ b/internal/wakeup/wakeup.go @@ -12,6 +12,11 @@ import ( "strings" ) +const ( + WriterBegin = "dfs-internal:writer-begin" + WriterEnd = "dfs-internal:writer-end" +) + type Listener struct { connection *net.UnixConn path string