fix(daemon/remote): a failed or interrupted bundle extract no longer destroys the work tree

internal/daemon/remote/bridge.go

@@ -100,14 +100,24 @@ func NewBridge(opts BridgeOptions) (*Bridge, error) {
 	if maxBundleBytes <= 0 {
 		maxBundleBytes = defaultMaxBundleBytes
 	}
+	bundleDir := strings.TrimSpace(opts.BundleDir)
+	if bundleDir != "" {
+		// A previous run may have died mid-swap, leaving a link's only tree in a
+		// staging dir. Nothing serves yet, so repair before the first upload.
+		recoverBundleDir(bundleDir, func(format string, args ...any) {
+			if opts.Log != nil {
+				opts.Log(fmt.Sprintf(format, args...))
+			}
+		})
+	}
 	return &Bridge{
 		server:           opts.Server,
 		auth:             opts.Authenticator,
 		attest:           attest,
 		minVersion:       minVersion,
 		handshakeTimeout: handshakeTimeout,
 		authFailDelay:    authFailDelay,
-		bundleDir:        strings.TrimSpace(opts.BundleDir),
+		bundleDir:        bundleDir,
 		maxBundleBytes:   maxBundleBytes,
 		log:              opts.Log,
 		sem:              make(chan struct{}, maxConns),

internal/daemon/remote/bundle.go

@@ -12,10 +12,13 @@ import (
 	"os"
 	"os/exec"
 	"path/filepath"
+	"slices"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/Gitlawb/zero/internal/daemon"
+	"github.com/Gitlawb/zero/internal/lockutil"
 )
 
 // gitTimeout bounds a single git invocation (bundle create/verify, clone) so a
@@ -129,16 +132,18 @@ func (b *Bridge) receiveBundle(conn net.Conn) bundleResult {
 		return bundleResult{Message: "stage bundle: " + err.Error()}
 	}
 
-	ctx, cancel := context.WithTimeout(context.Background(), gitTimeout)
-	defer cancel()
-	if err := gitBundleVerify(ctx, tmpName); err != nil {
+	verifyCtx, cancelVerify := context.WithTimeout(context.Background(), gitTimeout)
+	defer cancelVerify()
+	if err := gitBundleVerify(verifyCtx, tmpName); err != nil {
 		return bundleResult{Message: "bundle verify: " + err.Error()}
 	}
 	dest := filepath.Join(b.bundleDir, id)
 	if !withinDir(b.bundleDir, dest) {
 		return bundleResult{Message: "invalid link id"}
 	}
-	if err := extractBundle(ctx, tmpName, dest); err != nil {
+	// extractBundle starts the clone's own gitTimeout once it holds the lock for
+	// dest, so an upload queued behind another does not spend that budget waiting.
+	if err := extractBundle(context.Background(), tmpName, dest, b.logf); err != nil {
 		return bundleResult{Message: "extract bundle: " + err.Error()}
 	}
 	return bundleResult{OK: true, Path: dest}
@@ -168,27 +173,294 @@ func streamFramesToFile(r io.Reader, w io.Writer, size int64) error {
 	return nil
 }
 
-// extractBundle clones bundleFile into a staging dir, then atomically renames it
-// over dest (replacing any prior extraction for this link id). git clone needs a
-// non-existent target, so the staging+rename keeps the live dest intact on error.
-func extractBundle(ctx context.Context, bundleFile, dest string) error {
+// stagingPrefix names the per-extract staging directories created beside dest.
+// sanitizeLinkID refuses every dot-prefixed id so a link can never name one.
+const stagingPrefix = ".staging-"
+
+// lockDirName holds the per-link advisory lock files that serialize extracts
+// across processes. Dot-prefixed for the same reason stagingPrefix is.
+const lockDirName = ".extract-locks"
+
+// stagingLinkFile records, inside a staging dir, which link the backup beside it
+// belongs to. Without it a crash leaves an orphan nothing can attribute.
+const stagingLinkFile = "link"
+
+// extractLockPoll is how often a cross-process extract lock is retried.
+const extractLockPoll = 50 * time.Millisecond
+
+// renameDir moves a directory into its published location. It is a var so tests
+// can force a failure at the steps whose errors would otherwise be unrecoverable.
+var renameDir = os.Rename
+
+// extractLocks serializes extracts per destination. Each bundle upload runs in
+// its own connection goroutine, so two uploads of one link id would otherwise
+// interleave their swap steps and clobber each other.
+var extractLocks = struct {
+	mu    sync.Mutex
+	locks map[string]*extractLock
+}{locks: map[string]*extractLock{}}
+
+type extractLock struct {
+	mu   sync.Mutex
+	refs int
+}
+
+// lockExtract blocks until dest is free and returns its release func. Entries
+// are refcounted so the map cannot grow with every link id ever uploaded.
+func lockExtract(dest string) func() {
+	extractLocks.mu.Lock()
+	entry := extractLocks.locks[dest]
+	if entry == nil {
+		entry = &extractLock{}
+		extractLocks.locks[dest] = entry
+	}
+	entry.refs++
+	extractLocks.mu.Unlock()
+
+	entry.mu.Lock()
+	return func() {
+		entry.mu.Unlock()
+		extractLocks.mu.Lock()
+		entry.refs--
+		if entry.refs == 0 {
+			delete(extractLocks.locks, dest)
+		}
+		extractLocks.mu.Unlock()
+	}
+}
+
+// lockExtractFile takes the cross-process advisory lock for dest, waiting until
+// ctx is done or the wait budget runs out. The in-process lock already excludes
+// this daemon's own goroutines; this excludes a second daemon sharing the dir.
+func lockExtractFile(ctx context.Context, bundleDir, dest string) (func(), error) {
+	deadline := time.NewTimer(gitTimeout)
+	defer deadline.Stop()
+	for {
+		release, held, err := tryLockExtractFile(bundleDir, dest)
+		if err != nil {
+			return nil, err
+		}
+		if !held {
+			return release, nil
+		}
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case <-deadline.C:
+			return nil, fmt.Errorf("remote: timed out waiting for the extract lock on %s", dest)
+		case <-time.After(extractLockPoll):
+		}
+	}
+}
+
+// tryLockExtractFile takes the per-link advisory lock without waiting. It
+// reports held when a live extract owns the link, which is never an error: the
+// caller either waits or leaves that link alone.
+func tryLockExtractFile(bundleDir, dest string) (release func(), held bool, err error) {
+	lockDir := filepath.Join(bundleDir, lockDirName)
+	if err := os.MkdirAll(lockDir, 0o700); err != nil {
+		return nil, false, err
+	}
+	lock, err := lockutil.TryAcquireFileLockAt(bundleDir, filepath.Join(lockDir, filepath.Base(dest)+".lock"))
+	if err != nil {
+		if errors.Is(err, lockutil.ErrLockHeld) {
+			return nil, true, nil
+		}
+		return nil, false, err
+	}
+	return func() { _ = lock.Release() }, false, nil
+}
+
+// recoverBundleDir repairs what a crash left behind in dir. A staging dir whose
+// backup belongs to a link with no live tree is put back; one that no extract
+// can still own is removed. It is called once at bridge construction, before any
+// upload is served, and never removes a staging dir a live extract may hold.
+func recoverBundleDir(dir string, logf func(string, ...any)) {
+	if logf == nil {
+		logf = func(string, ...any) {}
+	}
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		if !os.IsNotExist(err) {
+			logf("remote: could not scan bundle dir %s: %v", dir, err)
+		}
+		return
+	}
+	// One link can have several staged backups: a cleanup that could not finish
+	// leaves one behind, and a later crash adds another. Newest first, so the
+	// tree that comes back is the most recent one rather than whichever the
+	// directory happened to list first.
+	staged := make([]string, 0, len(entries))
+	backupTime := map[string]time.Time{}
+	for _, entry := range entries {
+		if !entry.IsDir() || !strings.HasPrefix(entry.Name(), stagingPrefix) {
+			continue
+		}
+		path := filepath.Join(dir, entry.Name())
+		staged = append(staged, path)
+		if info, err := os.Stat(filepath.Join(path, "backup")); err == nil {
+			backupTime[path] = info.ModTime()
+		}
+	}
+	slices.SortFunc(staged, func(a, b string) int {
+		return backupTime[b].Compare(backupTime[a])
+	})
+
+	for _, staging := range staged {
+		if restoreStagedBackup(dir, staging, logf) {
+			continue
+		}
+		// No backup to attribute. A dir with a .git at its root is not staging at
+		// all: link ids starting with '.' used to be accepted, so this may be a
+		// work tree someone published under a name that now looks reserved.
+		// Never reap that.
+		if _, err := os.Stat(filepath.Join(staging, ".git")); err == nil {
+			logf("remote: %s holds a work tree, not a staged extract; leaving it in place", staging)
+			continue
+		}
+		// Only reap once no clone can still be running: gitTimeout bounds a
+		// clone, so anything older than that is abandoned.
+		info, err := os.Stat(staging)
+		if err != nil || time.Since(info.ModTime()) < 2*gitTimeout {
+			continue
+		}
+		if err := os.RemoveAll(staging); err != nil {
+			logf("remote: could not remove abandoned staging dir %s: %v", staging, err)
+		}
+	}
+}
+
+// restoreStagedBackup puts a staged backup back if its link has no live tree.
+// It reports whether staging was dealt with and needs no further handling.
+func restoreStagedBackup(dir, staging string, logf func(string, ...any)) bool {
+	backup := filepath.Join(staging, "backup")
+	if _, err := os.Stat(backup); err != nil {
+		return false
+	}
+	raw, err := os.ReadFile(filepath.Join(staging, stagingLinkFile))
+	if err != nil {
+		logf("remote: staged tree in %s has no link marker; leaving it in place", staging)
+		return true
+	}
+	id, err := sanitizeLinkID(string(raw))
+	if err != nil {
+		logf("remote: staged tree in %s names an invalid link (%v); leaving it in place", staging, err)
+		return true
+	}
+	dest := filepath.Join(dir, id)
+	if !withinDir(dir, dest) {
+		logf("remote: staged tree in %s names a link outside the bundle dir; leaving it in place", staging)
+		return true
+	}
+	// A live extract mid-swap looks exactly like a crashed one: its backup is
+	// aside and dest is briefly absent. Only the lock tells them apart, so skip
+	// any link something still owns rather than taking its tree.
+	release, held, err := tryLockExtractFile(dir, dest)
+	if err != nil {
+		logf("remote: could not lock %s while recovering %s: %v", id, staging, err)
+		return true
+	}
+	if held {
+		return true
+	}
+	defer release()
+	if destInfo, err := os.Stat(dest); err == nil {
+		// The link already has a tree. Only drop the backup when it is provably
+		// the older copy; otherwise it may be the newer one a restart has not
+		// published yet, and deleting it would lose that work.
+		backupInfo, statErr := os.Stat(backup)
+		if statErr != nil || !backupInfo.ModTime().Before(destInfo.ModTime()) {
+			logf("remote: staged tree in %s is not older than the live tree for %s; leaving it in place", staging, id)
+			return true
+		}
+		if err := os.RemoveAll(staging); err != nil {
+			logf("remote: could not remove superseded staging dir %s: %v", staging, err)
+		}
+		return true
+	}
+	if err := os.Rename(backup, dest); err != nil {
+		logf("remote: could not restore the staged tree for %s from %s: %v", id, staging, err)
+		return true
+	}
+	logf("remote: restored the work tree for %s from %s after an interrupted extract", id, staging)
+	if err := os.RemoveAll(staging); err != nil {
+		logf("remote: could not remove staging dir %s after restoring %s: %v", staging, id, err)
+	}
+	return true
+}
+
+// extractBundle clones bundleFile into a staging dir beside dest, then swaps the
+// clone into place (replacing any prior extraction for this link id). git clone
+// needs a non-existent target, hence the staging dir. The live tree is moved
+// aside rather than deleted and is put back if the publish fails, so on every
+// error return dest holds either the prior extraction or the new one, never
+// neither. Swapping a directory is two renames and cannot be made atomic, so a
+// crash between them leaves dest absent with the prior tree in staging/backup;
+// nothing reaps that on restart. logf may be nil.
+func extractBundle(ctx context.Context, bundleFile, dest string, logf func(string, ...any)) error {
+	if logf == nil {
+		logf = func(string, ...any) {}
+	}
 	parent := filepath.Dir(dest)
 	if err := os.MkdirAll(parent, 0o700); err != nil {
 		return err
 	}
-	staging, err := os.MkdirTemp(parent, ".staging-*")
+	unlock := lockExtract(dest)
+	defer unlock()
+	unlockFile, err := lockExtractFile(ctx, parent, dest)
 	if err != nil {
 		return err
 	}
-	defer func() { _ = os.RemoveAll(staging) }()
+	defer unlockFile()
+
+	staging, err := os.MkdirTemp(parent, stagingPrefix+"*")
+	if err != nil {
+		return err
+	}
+	// Staging also holds the prior tree while the swap is in flight, so it is
+	// only cleaned up once dest is known to hold one of the two trees.
+	cleanupStaging := true
+	defer func() {
+		if !cleanupStaging {
+			return
+		}
+		// A failure here strands a whole copy of the prior tree under a
+		// dot-prefixed dir nothing else enumerates, so say so rather than
+		// leaking it silently.
+		if err := os.RemoveAll(staging); err != nil {
+			logf("remote: could not remove bundle staging dir %s: %v", staging, err)
+		}
+	}()
+	cloneCtx, cancelClone := context.WithTimeout(ctx, gitTimeout)
+	defer cancelClone()
 	cloneDest := filepath.Join(staging, "repo")
-	if err := gitClone(ctx, bundleFile, cloneDest); err != nil {
+	if err := gitClone(cloneCtx, bundleFile, cloneDest); err != nil {
+		return err
+	}
+
+	// Every rename stays inside parent, so none of them crosses a filesystem.
+	backup := filepath.Join(staging, "backup")
+	// Record the link before moving its tree, so a crash in the swap window
+	// leaves something recoverBundleDir can attribute and put back.
+	if err := os.WriteFile(filepath.Join(staging, stagingLinkFile), []byte(filepath.Base(dest)), 0o600); err != nil {
 		return err
 	}
-	if err := os.RemoveAll(dest); err != nil {
+	restore := func() error { return nil }
+	if err := os.Rename(dest, backup); err == nil {
+		restore = func() error { return renameDir(backup, dest) }
+	} else if !os.IsNotExist(err) {
 		return err
 	}
-	return os.Rename(cloneDest, dest)
+	if err := renameDir(cloneDest, dest); err != nil {
+		if restoreErr := restore(); restoreErr != nil {
+			// dest is empty and the only copy of the prior tree is the backup,
+			// so keep staging rather than deleting the tree on the way out.
+			cleanupStaging = false
+			return fmt.Errorf("publish extraction: %w (prior tree left in %s: %v)", err, backup, restoreErr)
+		}
+		return err
+	}
+	return nil
 }
 
 // ---- client side -----------------------------------------------------------
@@ -323,8 +595,9 @@ func runGit(ctx context.Context, dir string, args ...string) error {
 }
 
 // sanitizeLinkID validates a link id used as a single path component under the
-// bundle dir. It allows letters, digits, '-', '_', '.', forbids the traversal
-// names, and caps the length — so it can never escape the bundle dir.
+// bundle dir. It allows letters, digits, '-', '_', '.', forbids a leading '.',
+// and caps the length, so an id can never escape the bundle dir and can never
+// name one of the stagingPrefix directories an extract creates beside it.
 func sanitizeLinkID(id string) (string, error) {
 	id = strings.TrimSpace(id)
 	if id == "" {
@@ -333,8 +606,8 @@ func sanitizeLinkID(id string) (string, error) {
 	if len(id) > 128 {
 		return "", errors.New("remote: link id too long (max 128)")
 	}
-	if id == "." || id == ".." {
-		return "", errors.New("remote: invalid link id")
+	if strings.HasPrefix(id, ".") {
+		return "", errors.New("remote: link id may not start with '.'")
 	}
 	for _, r := range id {
 		switch {