Expire idle snapshots and abandoned jobs, cap snapshot count

A background sweep every minute drops snapshots unused for an hour,
finished jobs, and uploads never committed, deleting their workspaces.
At most 8 snapshots are held; the least recently used is evicted.

indexer.go

@@ -11,6 +11,7 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+	"time"
 	"unicode/utf8"
 )
 
@@ -54,6 +55,7 @@ type Snapshot struct {
 	References       int
 	Warnings         []string
 	Root             string
+	lastUsed         time.Time
 	delivered        map[uint32][][2]int
 	released         map[uint32]bool
 }

main.go

@@ -39,9 +39,15 @@ type Job struct {
 	Error      *string  `json:"error"`
 	Warnings   []string `json:"warnings"`
 	root       string
-	owner      string // session that created the job
+	owner      string    // session that created the job
+	updated    time.Time // last progress, used to expire abandoned jobs
 }
 
+const (
+	idleExpiry   = time.Hour // snapshots unused and jobs untouched for this long are dropped
+	maxSnapshots = 8         // ponytail: fixed cap sized for the 2GB container; make it configurable if memory changes
+)
+
 type Server struct {
 	jobsMu      sync.Mutex
 	jobs        map[string]*Job
@@ -76,12 +82,18 @@ func main() {
 	open := flag.Bool("open", false, "open the viewer in a browser")
 	flag.Parse()
 
+	handler := NewServer(filepath.Join(os.TempDir(), "codenavigator"))
 	server := &http.Server{
 		Addr:              fmt.Sprintf("0.0.0.0:%d", port),
-		Handler:           NewServer(filepath.Join(os.TempDir(), "codenavigator")),
+		Handler:           handler,
 		ReadHeaderTimeout: 30 * time.Second,
 		IdleTimeout:       2 * time.Minute,
 	}
+	go func() {
+		for now := range time.Tick(time.Minute) {
+			handler.expire(now)
+		}
+	}()
 	log.Printf("CodeNavigator listening on http://127.0.0.1:%d", port)
 	if *open {
 		_ = exec.Command("xdg-open", fmt.Sprintf("http://127.0.0.1:%d", port)).Start()
@@ -145,7 +157,7 @@ func (s *Server) createJob(w http.ResponseWriter, r *http.Request) {
 		writeError(w, 500, err.Error())
 		return
 	}
-	job := &Job{ID: id, SourceKind: request.Kind, Name: "Local codebase", Phase: "receiving", Message: "Waiting for files…", Warnings: []string{}, root: root, owner: owner}
+	job := &Job{ID: id, SourceKind: request.Kind, Name: "Local codebase", Phase: "receiving", Message: "Waiting for files…", Warnings: []string{}, root: root, owner: owner, updated: time.Now()}
 	if request.Name != nil {
 		job.Name = *request.Name
 	}
@@ -210,6 +222,7 @@ func (s *Server) jobRoute(w http.ResponseWriter, r *http.Request, parts []string
 			job.Completed++
 			job.Total = job.Completed
 			job.Message = fmt.Sprintf("Received %d files", job.Completed)
+			job.updated = time.Now()
 		}
 		s.jobsMu.Unlock()
 		writeJSON(w, 200, map[string]bool{"ok": true})
@@ -258,6 +271,7 @@ func (s *Server) snapshotRoute(w http.ResponseWriter, r *http.Request, parts []s
 		writeError(w, 404, "Unknown snapshot")
 		return
 	}
+	snapshot.lastUsed = time.Now()
 	switch action {
 	case "":
 		body := map[string]any{
@@ -528,7 +542,18 @@ func (s *Server) indexJob(id, root string) {
 			delete(s.snapshots, key)
 		}
 	}
+	snapshot.lastUsed = time.Now()
 	s.snapshots[snapshot.ID] = snapshot
+	for len(s.snapshots) > maxSnapshots {
+		oldest := snapshot
+		for _, candidate := range s.snapshots {
+			if candidate.lastUsed.Before(oldest.lastUsed) {
+				oldest = candidate
+			}
+		}
+		oldRoots = append(oldRoots, oldest.Root)
+		delete(s.snapshots, oldest.ID)
+	}
 	s.snapshotsMu.Unlock()
 	for _, old := range oldRoots {
 		if old != root {
@@ -539,17 +564,48 @@ func (s *Server) indexJob(id, root string) {
 	s.jobsMu.Lock()
 	if job := s.jobs[id]; job != nil {
 		job.Phase, job.Message = "ready", "Landscape ready"
+		job.updated = time.Now()
 		job.SnapshotID = &snapshot.ID
 		job.Completed = max(job.Total, job.Completed)
 	}
 	s.jobsMu.Unlock()
 }
 
+// expire drops snapshots idle since before now-idleExpiry, finished jobs, and uploads
+// that were abandoned before commit, deleting their workspaces.
+func (s *Server) expire(now time.Time) {
+	cutoff := now.Add(-idleExpiry)
+	var roots []string
+	s.snapshotsMu.Lock()
+	for id, snapshot := range s.snapshots {
+		if snapshot.lastUsed.Before(cutoff) {
+			roots = append(roots, snapshot.Root)
+			delete(s.snapshots, id)
+		}
+	}
+	s.snapshotsMu.Unlock()
+	s.jobsMu.Lock()
+	for id, job := range s.jobs {
+		finished := job.SnapshotID != nil || job.Error != nil
+		if job.updated.Before(cutoff) && (finished || job.Phase == "receiving") {
+			if job.Phase == "receiving" {
+				roots = append(roots, job.root)
+			}
+			delete(s.jobs, id)
+		}
+	}
+	s.jobsMu.Unlock()
+	for _, root := range roots {
+		s.removeWorkspace(root)
+	}
+}
+
 func (s *Server) updateJob(id, phase string, completed, total int, message string) {
 	s.jobsMu.Lock()
 	defer s.jobsMu.Unlock()
 	if job := s.jobs[id]; job != nil {
 		job.Phase, job.Completed, job.Total, job.Message = phase, completed, total, message
+		job.updated = time.Now()
 	}
 }
 
@@ -558,6 +614,7 @@ func (s *Server) failJob(id, message string) {
 	defer s.jobsMu.Unlock()
 	if job := s.jobs[id]; job != nil {
 		job.Phase, job.Message, job.Error = "error", "Indexing failed", &message
+		job.updated = time.Now()
 	}
 }
 

main_test.go

@@ -13,6 +13,7 @@ import (
 	"reflect"
 	"strings"
 	"testing"
+	"time"
 )
 
 func TestDeliveryRangesMergeBeforeSourceIsReleased(t *testing.T) {
@@ -323,3 +324,40 @@ func TestSessionsKeepIndependentSnapshots(t *testing.T) {
 		t.Fatalf("snapshots = %d, want 2", len(server.snapshots))
 	}
 }
+
+func TestExpireDropsIdleSnapshotsAndAbandonedUploads(t *testing.T) {
+	server, c := newTestServer(t)
+	_, snapshot := c.indexLocal(map[string]string{"a.go": "package a"})
+	var pending struct{ JobID string }
+	c.json("POST", "/api/jobs", `{"kind":"local"}`, 201, &pending)
+	c.json("POST", "/api/jobs/"+pending.JobID+"/files?path=b.go", "package b", 200, nil)
+	snapshotRoot := server.snapshots[snapshot].Root
+	pendingRoot := server.jobs[pending.JobID].root
+
+	server.expire(time.Now())
+	c.json("GET", "/api/snapshots/"+snapshot, "", 200, nil)
+
+	server.expire(time.Now().Add(idleExpiry + time.Minute))
+	c.json("GET", "/api/snapshots/"+snapshot, "", 404, nil)
+	c.json("GET", "/api/jobs/"+pending.JobID, "", 404, nil)
+	for _, root := range []string{snapshotRoot, pendingRoot} {
+		if _, err := os.Stat(root); !os.IsNotExist(err) {
+			t.Errorf("workspace %s should be deleted", root)
+		}
+	}
+	if len(server.jobs) != 0 {
+		t.Errorf("jobs left: %d", len(server.jobs))
+	}
+}
+
+func TestSnapshotCapEvictsLeastRecentlyUsed(t *testing.T) {
+	server, first := newTestServer(t)
+	_, oldest := first.indexLocal(map[string]string{"a.go": "package a"})
+	for i := range maxSnapshots {
+		first.stranger().indexLocal(map[string]string{"a.go": fmt.Sprint("package p", i)})
+	}
+	if len(server.snapshots) != maxSnapshots {
+		t.Fatalf("snapshots = %d, want %d", len(server.snapshots), maxSnapshots)
+	}
+	first.json("GET", "/api/snapshots/"+oldest, "", 404, nil)
+}