simplify runtime startup and watcher

elithrar · elithrar · commit 43ec7122ca44 · 2026-04-06T08:33:42.000-04:00
diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go
@@ -45,13 +45,14 @@ type mountFailure struct {
 
 type repoRuntime struct {
 	cfg      model.RepoConfig
+	ctx      context.Context
+	cancel   context.CancelFunc
 	snapshot *snapshot.Store
 	overlay  *overlay.Store
 	hydrator *hydrator.Service
 	resolver *fusefs.Resolver
 	mfs      fusefs.MountedFS
 	state    model.RepoRuntimeState
-	stop     chan struct{}
 }
 
 type aheadBehind struct {
@@ -309,53 +310,68 @@ func (s *Service) Unmount(ctx context.Context, name string) error {
 // start a FUSE mount or any background goroutines, so it's safe to call from
 // short-lived CLI commands like add-repo.
 func (s *Service) prepareRepo(ctx context.Context, cfg model.RepoConfig) error {
-	if err := os.MkdirAll(cfg.MountPath, 0o755); err != nil {
-		return err
-	}
-	if err := s.git.CloneBlobless(ctx, cfg); err != nil {
-		return err
-	}
-	snap, err := snapshot.New(ctx, cfg.MetaDBPath)
+	snap, _, _, _, err := s.ensurePreparedRepo(ctx, cfg)
 	if err != nil {
 		return err
 	}
 	defer snap.Close()
-	_, err = s.publishHeadSnapshot(ctx, cfg, snap)
-	return err
+	return nil
 }
 
-// mountRepo opens all stores, starts the FUSE server, watcher, and refresh
-// loop. Called by the daemon's Start for each registered repo.
-func (s *Service) mountRepo(ctx context.Context, cfg model.RepoConfig) error {
+// ensurePreparedRepo makes sure the repo is cloned and has an initial snapshot.
+// The returned snapshot store remains open for callers that need to continue
+// into runtime startup.
+func (s *Service) ensurePreparedRepo(ctx context.Context, cfg model.RepoConfig) (*snapshot.Store, string, string, int64, error) {
 	if err := os.MkdirAll(cfg.MountPath, 0o755); err != nil {
-		return err
+		return nil, "", "", 0, err
 	}
-	// Clone if not already present (idempotent)
 	if err := s.git.CloneBlobless(ctx, cfg); err != nil {
-		return err
+		return nil, "", "", 0, err
 	}
 	headOID, headRef, err := s.git.ResolveHEAD(ctx, cfg)
 	if err != nil {
-		return err
+		return nil, "", "", 0, err
 	}
 	snap, err := snapshot.New(ctx, cfg.MetaDBPath)
 	if err != nil {
-		return err
+		return nil, "", "", 0, err
 	}
-	gen, err := snap.CurrentGeneration(ctx)
-	if err != nil || gen == 0 {
-		var bErr error
-		gen, _, bErr = s.publishSnapshot(ctx, cfg, snap, headOID, headRef)
-		if bErr != nil {
+	storedOID, storedRef, gen, err := snap.ReadState(ctx)
+	if err != nil || gen == 0 || storedOID != headOID || storedRef != headRef {
+		gen, _, err = s.publishSnapshot(ctx, cfg, snap, headOID, headRef)
+		if err != nil {
 			snap.Close()
-			return bErr
+			return nil, "", "", 0, err
 		}
 	}
+	return snap, headOID, headRef, gen, nil
+}
+
+// mountRepo opens all stores, starts the FUSE server, watcher, and refresh
+// loop. Called by the daemon's Start for each registered repo.
+func (s *Service) mountRepo(ctx context.Context, cfg model.RepoConfig) error {
+	snap, headOID, headRef, gen, err := s.ensurePreparedRepo(ctx, cfg)
+	if err != nil {
+		return err
+	}
 	ov, err := overlay.New(ctx, cfg)
 	if err != nil {
 		snap.Close()
 		return err
 	}
+	baseLookup := func(path string) (model.BaseNode, bool) {
+		return snap.GetNode(gen, path)
+	}
+	if err := ov.Reconcile(ctx, baseLookup); err != nil {
+		ov.Close()
+		snap.Close()
+		return err
+	}
+	if err := s.git.ReadTreeHEAD(ctx, cfg); err != nil {
+		ov.Close()
+		snap.Close()
+		return err
+	}
 	h := hydrator.New(s.git)
 
 	resolver := &fusefs.Resolver{Snapshot: snap, Overlay: ov}
@@ -378,18 +394,20 @@ func (s *Service) mountRepo(ctx context.Context, cfg model.RepoConfig) error {
 		s.logger.Error("fuse mount failed, running without FUSE", "repo", cfg.Name, "error", err)
 		mfs = nil
 	}
+	runtimeCtx, cancel := context.WithCancel(ctx)
 
 	rt := &repoRuntime{
 		cfg:      cfg,
+		ctx:      runtimeCtx,
+		cancel:   cancel,
 		snapshot: snap,
 		overlay:  ov,
 		hydrator: h,
 		resolver: resolver,
 		mfs:      mfs,
 		state:    newRuntimeState(cfg.ID, headOID, headRef, gen),
-		stop:     make(chan struct{}),
 	}
-	s.startRuntime(ctx, rt)
+	s.startRuntime(rt)
 
 	return nil
 }
@@ -402,8 +420,18 @@ func (s *Service) onHEADChanged(ctx context.Context, rt *repoRuntime) {
 	}
 	s.mu.Lock()
 	prevOID := rt.state.CurrentHEADOID
+	prevRef := rt.state.CurrentHEADRef
 	s.mu.Unlock()
 	if oid == prevOID {
+		if ref == prevRef {
+			return
+		}
+		if err := rt.snapshot.UpdateHEADRef(ctx, ref); err != nil {
+			s.logger.Warn("snapshot head_ref update failed", "repo", rt.cfg.Name, "error", err)
+		}
+		s.mu.Lock()
+		rt.state.CurrentHEADRef = ref
+		s.mu.Unlock()
 		return
 	}
 	gen, phase, err := s.publishSnapshot(ctx, rt.cfg, rt.snapshot, oid, ref)
@@ -444,10 +472,10 @@ func (s *Service) refreshLoop(rt *repoRuntime) {
 	defer ticker.Stop()
 	for {
 		select {
-		case <-rt.stop:
+		case <-rt.ctx.Done():
 			return
 		case <-ticker.C:
-			ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+			ctx, cancel := context.WithTimeout(rt.ctx, 30*time.Second)
 			err := s.git.Fetch(ctx, rt.cfg)
 			if err != nil {
 				s.mu.Lock()
@@ -510,15 +538,6 @@ func (s *Service) readPersistedStatus(ctx context.Context, cfg model.RepoConfig)
 	return st
 }
 
-func (s *Service) publishHeadSnapshot(ctx context.Context, cfg model.RepoConfig, snap *snapshot.Store) (int64, error) {
-	oid, ref, err := s.git.ResolveHEAD(ctx, cfg)
-	if err != nil {
-		return 0, err
-	}
-	gen, _, err := s.publishSnapshot(ctx, cfg, snap, oid, ref)
-	return gen, err
-}
-
 func (s *Service) publishSnapshot(ctx context.Context, cfg model.RepoConfig, snap *snapshot.Store, oid string, ref string) (int64, string, error) {
 	nodes, err := s.git.BuildTreeIndex(ctx, cfg, oid)
 	if err != nil {
@@ -547,23 +566,21 @@ func (s *Service) fetchState(ctx context.Context, cfg model.RepoConfig) (aheadBe
 	return aheadBehind{ahead: ahead, behind: behind, diverged: diverged}, nil
 }
 
-func (s *Service) startRuntime(ctx context.Context, rt *repoRuntime) {
+func (s *Service) startRuntime(rt *repoRuntime) {
 	s.mu.Lock()
 	s.running[rt.cfg.ID] = rt
 	s.mu.Unlock()
 
 	go s.refreshLoop(rt)
 
 	w := watcher.New(500 * time.Millisecond)
-	go w.Watch(ctx, rt.cfg.GitDir, func(sig watcher.Signal) {
-		if sig.HEADChanged {
-			s.onHEADChanged(ctx, rt)
-		}
+	go w.Watch(rt.ctx, rt.cfg.GitDir, func() {
+		s.onHEADChanged(rt.ctx, rt)
 	})
 
 	if rt.mfs != nil {
 		go func() {
-			_ = rt.mfs.Join(ctx)
+			_ = rt.mfs.Join(rt.ctx)
 		}()
 	}
 }
@@ -654,7 +671,9 @@ func (s *Service) unmount(id model.RepoID) {
 }
 
 func (s *Service) stopRuntime(rt *repoRuntime) {
-	close(rt.stop)
+	if rt.cancel != nil {
+		rt.cancel()
+	}
 	if rt.mfs != nil {
 		_ = rt.mfs.Unmount()
 	}
diff --git a/internal/fusefs/merged.go b/internal/fusefs/merged.go
@@ -34,15 +34,10 @@ func childName(parent, entryPath string) (string, bool) {
 	return rel, true
 }
 
-type SnapshotLookup interface {
-	GetNode(generation int64, path string) (model.BaseNode, bool)
-	ListChildren(generation int64, parentPath string) ([]model.BaseNode, error)
-}
-
 type Resolver struct {
 	generation atomic.Int64
 	commitTime atomic.Int64 // unix seconds of HEAD commit
-	Snapshot   SnapshotLookup
+	Snapshot   model.SnapshotStore
 	Overlay    model.OverlayStore
 }
 
diff --git a/internal/fusefs/merged_test.go b/internal/fusefs/merged_test.go
@@ -14,6 +14,10 @@ type fakeSnapshot struct {
 	kids  map[string][]model.BaseNode
 }
 
+func (f *fakeSnapshot) PublishGeneration(_ context.Context, _ string, _ string, _ []model.BaseNode) (int64, error) {
+	return 0, nil
+}
+
 func (f *fakeSnapshot) GetNode(_ int64, path string) (model.BaseNode, bool) {
 	n, ok := f.nodes[path]
 	return n, ok
diff --git a/internal/overlay/store.go b/internal/overlay/store.go
@@ -46,9 +46,6 @@ func New(ctx context.Context, cfg model.RepoConfig) (*Store, error) {
 	if err := os.MkdirAll(upperDir, 0o755); err != nil {
 		return nil, err
 	}
-	if err := os.MkdirAll(filepath.Join(cfg.OverlayDir, "whiteouts"), 0o755); err != nil {
-		return nil, err
-	}
 	return &Store{db: db, repo: cfg, upperDir: upperDir}, nil
 }
 
diff --git a/internal/snapshot/store.go b/internal/snapshot/store.go
@@ -120,6 +120,11 @@ func (s *Store) ReadState(ctx context.Context) (headOID, headRef string, generat
 	return headOID, headRef, gen, nil
 }
 
+func (s *Store) UpdateHEADRef(ctx context.Context, ref string) error {
+	_, err := s.db.ExecContext(ctx, `INSERT INTO repo_state(key,value) VALUES(?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, "head_ref", ref)
+	return err
+}
+
 func (s *Store) GetNode(generation int64, path string) (model.BaseNode, bool) {
 	// Uses background context for backward compat; callers with a deadline
 	// should use GetNodeCtx.
diff --git a/internal/watcher/watcher.go b/internal/watcher/watcher.go
@@ -4,46 +4,84 @@ import (
 	"context"
 	"os"
 	"path/filepath"
+	"strings"
 	"time"
 )
 
-type Signal struct {
-	HEADChanged  bool
-	RefsChanged  bool
-	IndexChanged bool
-}
-
 type Poller struct {
-	interval time.Duration
-	prev     map[string]time.Time
+	interval        time.Duration
+	prev            map[string]time.Time
+	firstSeenChange map[string]struct{}
 }
 
 func New(interval time.Duration) *Poller {
 	if interval <= 0 {
 		interval = 500 * time.Millisecond
 	}
-	return &Poller{interval: interval, prev: map[string]time.Time{}}
+	return &Poller{interval: interval, prev: map[string]time.Time{}, firstSeenChange: map[string]struct{}{}}
 }
 
-func (p *Poller) Watch(ctx context.Context, gitDir string, fn func(Signal)) {
+func (p *Poller) Watch(ctx context.Context, gitDir string, fn func()) {
+	headPath := filepath.Join(gitDir, "HEAD")
 	t := time.NewTicker(p.interval)
 	defer t.Stop()
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		case <-t.C:
-			s := Signal{}
-			s.HEADChanged = p.changed(filepath.Join(gitDir, "HEAD"))
-			s.IndexChanged = p.changed(filepath.Join(gitDir, "index"))
-			s.RefsChanged = p.changed(filepath.Join(gitDir, "refs", "heads")) || p.changed(filepath.Join(gitDir, "refs", "remotes"))
-			if s.HEADChanged || s.RefsChanged || s.IndexChanged {
-				fn(s)
+			if p.headChanged(headPath) {
+				fn()
 			}
 		}
 	}
 }
 
+func (p *Poller) headChanged(headPath string) bool {
+	if p.changed(headPath) {
+		// A HEAD switch selects a different ref path. Prime that ref immediately
+		// so the next branch-tip advance is compared against the new baseline.
+		if refPath, ok := p.headRefPath(headPath); ok {
+			p.primeCurrentRef(refPath)
+		}
+		return true
+	}
+	refPath, ok := p.headRefPath(headPath)
+	if !ok {
+		return false
+	}
+	return p.currentRefChanged(refPath)
+}
+
+func (p *Poller) primeCurrentRef(refPath string) {
+	_ = p.currentRefChanged(refPath)
+}
+
+func (p *Poller) currentRefChanged(refPath string) bool {
+	if _, err := os.Stat(refPath); err != nil {
+		p.firstSeenChange[refPath] = struct{}{}
+		return false
+	}
+	return p.changed(refPath)
+}
+
+func (p *Poller) headRefPath(headPath string) (string, bool) {
+	data, err := os.ReadFile(headPath)
+	if err != nil {
+		return "", false
+	}
+	const prefix = "ref: "
+	line := strings.TrimSpace(string(data))
+	if !strings.HasPrefix(line, prefix) {
+		return "", false
+	}
+	ref := strings.TrimSpace(strings.TrimPrefix(line, prefix))
+	if ref == "" {
+		return "", false
+	}
+	return filepath.Join(filepath.Dir(headPath), filepath.FromSlash(ref)), true
+}
+
 func (p *Poller) changed(path string) bool {
 	st, err := os.Stat(path)
 	if err != nil {
@@ -53,7 +91,12 @@ func (p *Poller) changed(path string) bool {
 	prev, ok := p.prev[path]
 	p.prev[path] = mtime
 	if !ok {
+		if _, changeOnFirstSeen := p.firstSeenChange[path]; changeOnFirstSeen {
+			delete(p.firstSeenChange, path)
+			return true
+		}
 		return false
 	}
+	delete(p.firstSeenChange, path)
 	return mtime.After(prev)
 }
diff --git a/internal/watcher/watcher_test.go b/internal/watcher/watcher_test.go

Original file line number	Diff line number	Diff line change
`@@ -46,9 +46,6 @@ func New(ctx context.Context, cfg model.RepoConfig) (*Store, error) {`
`46`	`46`	`if err := os.MkdirAll(upperDir, 0o755); err != nil {`
`47`	`47`	`return nil, err`
`48`	`48`	`}`
`49`		`- if err := os.MkdirAll(filepath.Join(cfg.OverlayDir, "whiteouts"), 0o755); err != nil {`
`50`		`- return nil, err`
`51`		`- }`
`52`	`49`	`return &Store{db: db, repo: cfg, upperDir: upperDir}, nil`
`53`	`50`	`}`
`54`	`51`