[scanner] fix(contribute): persist task leases so a hub restart doesn't revoke every in-flight task

changelog.d/fixed-5681-lease-restart-persistence.md

@@ -0,0 +1 @@
+- **Contribute: a hub restart no longer throws away every in-flight contributor task** (#5681). The C4 server-issued task-lease registry lived only in hub memory, so any restart — routine under self-upgrade rolls — emptied it: every relay that kept working through the brief disconnect was told `no active lease for this task` on reconnect, its agent was interrupted mid-turn, and the same issue was re-assigned to the same relay seconds later, restarting the work from zero. Leases are now persisted to a `task-leases.json` ledger in the same PVC-backed contributors dir as the cooldown/streak ledgers, written on every lease record/renew/revoke and reloaded (minus expired entries) at hub startup, so a reconnecting relay resumes exactly the task the hub assigned it. Persistence grants nothing new: a restored lease is the identical `{identity, task, repo, number, tier, generation, expiry}` tuple the hub issued, the C4 exact-match contract and #2568 generation fence apply unchanged, and a lease revoked before the restart stays revoked.

src/pkg/dashboard/contribute_lease_persist.go

@@ -0,0 +1,151 @@
+package dashboard
+
+// contribute_lease_persist.go — persist the C4 task-lease registry across hub
+// restarts (kubestellar/hive#5681).
+//
+// Leases lived only in the hub's memory, so a hub restart (routine under
+// self-upgrade rolls) emptied the registry. Every relay that was mid-task kept
+// working through the brief disconnect and re-asserted its task on reconnect,
+// but lookupLease could no longer match any server-issued lease: the relay was
+// told "no active lease for this task", the revoke path interrupted the agent
+// mid-turn, and seconds later the SAME issue was re-assigned to the SAME relay
+// as a fresh task — ownership was never in question, only the record of it.
+//
+// The registry is now a PVC-backed ledger in the same contributors dir as the
+// cooldown/streak/verdict ledgers, written on every lease mutation (record,
+// renew, revoke) and loaded at hub construction with expired entries dropped.
+// Persistence changes NOTHING about what a lease grants: a restored lease is
+// byte-for-byte the {identity, task, repo, number, tier, generation, expiry}
+// tuple the hub itself issued, so the C4 exact-match contract and the #2568
+// generation fence apply to it unchanged, and a lease revoked before the
+// restart is absent from the file and stays absent.
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"time"
+)
+
+// taskLeasesFileName is the on-disk ledger for server-issued task leases
+// (#5681). The full path comes from taskLeasesPath(), which honours
+// HIVE_CONTRIBUTORS_DIR exactly as the other contribute ledgers do — that is
+// what lets the persistence round-trip tests point it at a temp dir.
+const taskLeasesFileName = "task-leases.json"
+
+func taskLeasesPath() string {
+	return filepath.Join(getContributorsDir(), taskLeasesFileName)
+}
+
+// taskLeaseRecord is the JSON shape of one persisted lease. It carries exactly
+// the fields lookupLease matches on plus the expiry — nothing is reconstructed
+// at load time.
+type taskLeaseRecord struct {
+	Identity  string    `json:"identity"`
+	TaskID    string    `json:"task_id"`
+	Repo      string    `json:"repo"`
+	Number    int       `json:"number"`
+	Tier      string    `json:"tier"`
+	Gen       uint64    `json:"gen"`
+	ExpiresAt time.Time `json:"expires_at"`
+}
+
+func (h *ContributeWSHub) leasesFilePath() string {
+	if h != nil && h.leasesFile != "" {
+		return h.leasesFile
+	}
+	return taskLeasesPath()
+}
+
+// loadLeases restores the lease registry from disk at hub construction
+// (#5681), dropping entries that expired while the hub was down or that are
+// malformed. A restored lease behaves identically to one recordLease minted in
+// this process: it is renewed by task_progress, revoked on every release path,
+// and expires on the same leaseTTL clock.
+func (h *ContributeWSHub) loadLeases() {
+	if h != nil && !h.persistTaskLedgers {
+		return
+	}
+	data, err := os.ReadFile(h.leasesFilePath())
+	if err != nil {
+		return
+	}
+	var records []taskLeaseRecord
+	if json.Unmarshal(data, &records) != nil {
+		return
+	}
+	now := time.Now()
+	restored := 0
+	h.leaseMu.Lock()
+	if h.leases == nil {
+		h.leases = make(map[string]*taskLease)
+	}
+	for _, rec := range records {
+		if rec.Identity == "" || rec.TaskID == "" || rec.Gen == 0 {
+			continue
+		}
+		if now.After(rec.ExpiresAt) {
+			continue
+		}
+		h.leases[rec.Identity] = &taskLease{
+			identity:  rec.Identity,
+			taskID:    rec.TaskID,
+			repo:      rec.Repo,
+			number:    rec.Number,
+			tier:      rec.Tier,
+			gen:       rec.Gen,
+			expiresAt: rec.ExpiresAt,
+		}
+		restored++
+	}
+	h.leaseMu.Unlock()
+	if restored > 0 {
+		h.logger.Info("[contribute-ws] restored task leases", "count", restored)
+	}
+}
+
+// saveLeases writes the current lease registry to its ledger atomically
+// (tmp + rename, like every other contribute ledger). Called after every
+// lease mutation; expired entries are skipped so the file cannot accumulate
+// stale claims.
+func (h *ContributeWSHub) saveLeases() {
+	if h != nil && !h.persistTaskLedgers {
+		return
+	}
+	now := time.Now()
+	h.leaseMu.Lock()
+	records := make([]taskLeaseRecord, 0, len(h.leases))
+	for _, l := range h.leases {
+		if now.After(l.expiresAt) {
+			continue
+		}
+		records = append(records, taskLeaseRecord{
+			Identity:  l.identity,
+			TaskID:    l.taskID,
+			Repo:      l.repo,
+			Number:    l.number,
+			Tier:      l.tier,
+			Gen:       l.gen,
+			ExpiresAt: l.expiresAt,
+		})
+	}
+	h.leaseMu.Unlock()
+	data, err := json.Marshal(records)
+	if err != nil {
+		h.logger.Warn("[contribute-ws] task leases marshal failed", "error", err)
+		return
+	}
+	path := h.leasesFilePath()
+	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+		h.logger.Warn("[contribute-ws] task leases directory creation failed", "error", err)
+		return
+	}
+	tmpPath := path + ".tmp"
+	if err := os.WriteFile(tmpPath, data, 0o644); err != nil {
+		h.logger.Warn("[contribute-ws] task leases write failed", "error", err)
+		return
+	}
+	if err := os.Rename(tmpPath, path); err != nil {
+		h.logger.Warn("[contribute-ws] task leases rename failed", "error", err)
+	}
+}

src/pkg/dashboard/contribute_lease_persist_test.go

@@ -0,0 +1,186 @@
+package dashboard
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+// contribute_lease_persist_test.go covers kubestellar/hive#5681: a hub restart
+// must not make every in-flight contributor task unresumable. Leases lived only
+// in memory, so a restart emptied the registry, lookupLease rejected every
+// reconnecting relay ("no active lease for this task"), the agent was
+// interrupted mid-turn, and the SAME issue was re-assigned to the SAME relay
+// seconds later. The registry is now a PVC-backed ledger; these tests exercise
+// the restart round-trip and prove persistence grants nothing C4 forbids.
+
+// leasePersistHub builds a hub whose contribute ledgers live in dir, so a
+// second hub built over the same dir simulates a hub process restart.
+func leasePersistHub(t *testing.T, dir string) *ContributeWSHub {
+	t.Helper()
+	t.Setenv("HIVE_CONTRIBUTORS_DIR", dir)
+	redirectContributeWSDisk(t, dir)
+	return covK2HubIn(t, dir)
+}
+
+// covK2HubIn is covK2Hub with an explicit dir (covK2Hub only creates a fresh
+// temp dir when HIVE_CONTRIBUTORS_DIR is unset, which the caller controls).
+func covK2HubIn(t *testing.T, dir string) *ContributeWSHub {
+	t.Helper()
+	hub, _ := covK2Hub(t)
+	if got := hub.leasesFilePath(); filepath.Dir(got) != dir {
+		t.Fatalf("lease ledger path %q is not in the test dir %q", got, dir)
+	}
+	return hub
+}
+
+// TestLeasePersist_ResumeSurvivesHubRestart is the core #5681 regression: a
+// lease recorded before a restart must still be re-adoptable by the exact
+// {identity, task, repo, number, generation} tuple after it, within its window.
+func TestLeasePersist_ResumeSurvivesHubRestart(t *testing.T) {
+	dir := t.TempDir()
+
+	hub1 := leasePersistHub(t, dir)
+	now := time.Now()
+	hub1.recordLease("c-restart", "ct-5617", "myorg/repo1", 5617, "contributor", 75, now)
+
+	// "Restart": a brand-new hub over the same ledger dir.
+	hub2 := leasePersistHub(t, dir)
+	if hub2.lookupLease("c-restart", "ct-5617", "myorg/repo1", 5617, 75, now.Add(time.Minute)) == nil {
+		t.Fatalf("#5681: a lease issued before a hub restart could not be re-adopted "+
+			"after it — the relay would be told %q and its agent interrupted mid-turn",
+			"no active lease for this task")
+	}
+
+	// C4 exact-match still applies to a restored lease: wrong gen, task, repo,
+	// or identity must all be rejected.
+	if hub2.lookupLease("c-restart", "ct-5617", "myorg/repo1", 5617, 74, now.Add(time.Minute)) != nil {
+		t.Fatalf("C4: a restored lease matched a stale generation")
+	}
+	if hub2.lookupLease("c-restart", "ct-other", "myorg/repo1", 5617, 75, now.Add(time.Minute)) != nil {
+		t.Fatalf("C4: a restored lease matched a different task id")
+	}
+	if hub2.lookupLease("c-other", "ct-5617", "myorg/repo1", 5617, 75, now.Add(time.Minute)) != nil {
+		t.Fatalf("C4: a restored lease matched a different identity")
+	}
+}
+
+// TestLeasePersist_RevokedLeaseStaysRevokedAcrossRestart: a lease released
+// before the restart is absent from the ledger and must not be resurrected —
+// persistence must not reopen the resurrection window C4 closed.
+func TestLeasePersist_RevokedLeaseStaysRevokedAcrossRestart(t *testing.T) {
+	dir := t.TempDir()
+
+	hub1 := leasePersistHub(t, dir)
+	now := time.Now()
+	hub1.recordLease("c-done", "ct-done", "myorg/repo1", 7, "contributor", 3, now)
+	hub1.revokeLease("c-done", "ct-done")
+
+	hub2 := leasePersistHub(t, dir)
+	if hub2.lookupLease("c-done", "ct-done", "myorg/repo1", 7, 3, now.Add(time.Minute)) != nil {
+		t.Fatalf("C4: a lease revoked before a restart was resurrected from the ledger")
+	}
+}
+
+// TestLeasePersist_ExpiredLeaseNotRestored: a lease whose window elapsed while
+// the hub was down is dropped at load, exactly as lookupLease would drop it.
+func TestLeasePersist_ExpiredLeaseNotRestored(t *testing.T) {
+	dir := t.TempDir()
+
+	// Write a ledger with one already-expired lease, as if the hub had been
+	// down for longer than leaseTTL.
+	records := []taskLeaseRecord{{
+		Identity:  "c-stale",
+		TaskID:    "ct-stale",
+		Repo:      "myorg/repo1",
+		Number:    9,
+		Tier:      "contributor",
+		Gen:       4,
+		ExpiresAt: time.Now().Add(-time.Minute),
+	}}
+	data, err := json.Marshal(records)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(dir, taskLeasesFileName), data, 0o644); err != nil {
+		t.Fatal(err)
+	}
+
+	hub := leasePersistHub(t, dir)
+	if hub.lookupLease("c-stale", "ct-stale", "myorg/repo1", 9, 4, time.Now()) != nil {
+		t.Fatalf("an expired lease was restored from the ledger and re-adopted")
+	}
+	hub.leaseMu.Lock()
+	_, present := hub.leases["c-stale"]
+	hub.leaseMu.Unlock()
+	if present {
+		t.Fatalf("an expired lease was loaded into the registry instead of dropped")
+	}
+}
+
+// TestLeasePersist_RenewalExtendsWindowAcrossRestart: the #4260 renewal clock
+// must survive a restart too — a long-running task renewed past its original
+// assignment window is still re-adoptable after a restart.
+func TestLeasePersist_RenewalExtendsWindowAcrossRestart(t *testing.T) {
+	dir := t.TempDir()
+
+	hub1 := leasePersistHub(t, dir)
+	assigned := time.Now().Add(-2 * leaseTTL)
+	hub1.recordLease("c-long", "ct-long", "myorg/repo1", 42, "contributor", 9, assigned)
+	// The relay kept reporting progress; the last renewal was just now.
+	hub1.renewLease("c-long", "ct-long", time.Now())
+
+	hub2 := leasePersistHub(t, dir)
+	if hub2.lookupLease("c-long", "ct-long", "myorg/repo1", 42, 9, time.Now().Add(time.Minute)) == nil {
+		t.Fatalf("#5681/#4260: a renewed lease did not survive a restart with its "+
+			"renewed window — expiry must persist from the LAST renewal (assigned %v ago)",
+			2*leaseTTL)
+	}
+}
+
+// TestLeasePersist_MalformedAndPartialRecordsSkipped: garbage in the ledger
+// must not mint leases. Records missing the identity, task, or generation —
+// the fields a resume must prove possession of — are skipped at load.
+func TestLeasePersist_MalformedAndPartialRecordsSkipped(t *testing.T) {
+	dir := t.TempDir()
+
+	future := time.Now().Add(time.Hour)
+	records := []taskLeaseRecord{
+		{Identity: "", TaskID: "ct-a", Repo: "myorg/repo1", Number: 1, Gen: 2, ExpiresAt: future},
+		{Identity: "c-b", TaskID: "", Repo: "myorg/repo1", Number: 2, Gen: 3, ExpiresAt: future},
+		{Identity: "c-c", TaskID: "ct-c", Repo: "myorg/repo1", Number: 3, Gen: 0, ExpiresAt: future},
+		{Identity: "c-ok", TaskID: "ct-ok", Repo: "myorg/repo1", Number: 4, Gen: 5, ExpiresAt: future},
+	}
+	data, err := json.Marshal(records)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(dir, taskLeasesFileName), data, 0o644); err != nil {
+		t.Fatal(err)
+	}
+
+	hub := leasePersistHub(t, dir)
+	hub.leaseMu.Lock()
+	count := len(hub.leases)
+	hub.leaseMu.Unlock()
+	if count != 1 {
+		t.Fatalf("expected exactly 1 restored lease, got %d — partial records must be skipped", count)
+	}
+	if hub.lookupLease("c-ok", "ct-ok", "myorg/repo1", 4, 5, time.Now()) == nil {
+		t.Fatalf("the one well-formed lease was not restored")
+	}
+
+	// A corrupt (non-JSON) ledger is ignored entirely rather than crashing.
+	if err := os.WriteFile(filepath.Join(dir, taskLeasesFileName), []byte("{not json"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	hub2 := leasePersistHub(t, dir)
+	hub2.leaseMu.Lock()
+	count2 := len(hub2.leases)
+	hub2.leaseMu.Unlock()
+	if count2 != 0 {
+		t.Fatalf("a corrupt ledger produced %d leases", count2)
+	}
+}