From e2f6e03d7cca41e9b158b438b39bf9561a98523c Mon Sep 17 00:00:00 2001 From: sec-check Date: Wed, 2 Sep 2026 11:40:12 -0400 Subject: [PATCH] [scanner] fix(contribute): persist task leases so a hub restart doesn't revoke every in-flight task Leases (C4 server-issued task ownership records) lived only in hub memory. A hub restart emptied the registry, so every reconnecting relay's resume failed lookupLease, got 'no active lease for this task', had its agent interrupted mid-turn, and was handed the same issue back seconds later as a fresh task. Persist the registry to task-leases.json in the PVC-backed contributors dir (same pattern as the cooldown/streak/verdict ledgers): saved on every recordLease/renewLease/revokeLease, loaded at hub construction with expired and malformed entries dropped. A restored lease is byte-for-byte the tuple the hub issued, so the C4 exact-match contract and the #2568 generation fence are unchanged; revoked leases are absent from the ledger and stay revoked; expiry still runs on the #4260 renewal clock. Fixes #5681 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Signed-off-by: sec-check --- .../fixed-5681-lease-restart-persistence.md | 1 + src/pkg/dashboard/contribute_lease_persist.go | 151 ++++++++++++++ .../contribute_lease_persist_test.go | 186 ++++++++++++++++++ src/pkg/dashboard/contribute_ws.go | 16 ++ 4 files changed, 354 insertions(+) create mode 100644 changelog.d/fixed-5681-lease-restart-persistence.md create mode 100644 src/pkg/dashboard/contribute_lease_persist.go create mode 100644 src/pkg/dashboard/contribute_lease_persist_test.go diff --git a/changelog.d/fixed-5681-lease-restart-persistence.md b/changelog.d/fixed-5681-lease-restart-persistence.md new file mode 100644 index 000000000..ddeb4497c --- /dev/null +++ b/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. diff --git a/src/pkg/dashboard/contribute_lease_persist.go b/src/pkg/dashboard/contribute_lease_persist.go new file mode 100644 index 000000000..2400cc88d --- /dev/null +++ b/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) + } +} diff --git a/src/pkg/dashboard/contribute_lease_persist_test.go b/src/pkg/dashboard/contribute_lease_persist_test.go new file mode 100644 index 000000000..caa4faf00 --- /dev/null +++ b/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) + } +} diff --git a/src/pkg/dashboard/contribute_ws.go b/src/pkg/dashboard/contribute_ws.go index 15fab6388..47d6b5889 100644 --- a/src/pkg/dashboard/contribute_ws.go +++ b/src/pkg/dashboard/contribute_ws.go @@ -552,6 +552,7 @@ type ContributeWSHub struct { failedTasksFile string noPRStreaksFile string noWorkVerdictsFile string + leasesFile string asyncActivitySave bool persistActivity bool persistTaskLedgers bool @@ -672,6 +673,7 @@ func (h *ContributeWSHub) recordLease(identity, taskID, repo string, number int, expiresAt: now.Add(leaseTTL), } h.leaseMu.Unlock() + h.saveLeases() } // renewLease extends an identity's server-issued lease window when the relay proves @@ -693,10 +695,15 @@ func (h *ContributeWSHub) renewLease(identity, taskID string, now time.Time) { return } h.leaseMu.Lock() + renewed := false if l, ok := h.leases[identity]; ok && l.taskID == taskID { l.expiresAt = now.Add(leaseTTL) + renewed = true } h.leaseMu.Unlock() + if renewed { + h.saveLeases() + } } // revokeLease removes the server-authoritative lease for an identity on any release @@ -711,10 +718,15 @@ func (h *ContributeWSHub) revokeLease(identity, taskID string) { return } h.leaseMu.Lock() + revoked := false if l, ok := h.leases[identity]; ok && (taskID == "" || l.taskID == taskID) { delete(h.leases, identity) + revoked = true } h.leaseMu.Unlock() + if revoked { + h.saveLeases() + } } // lookupLease returns the active, unexpired server-issued lease for an identity that @@ -741,6 +753,8 @@ func (h *ContributeWSHub) lookupLease(identity, taskID, repo string, number int, } if now.After(l.expiresAt) { // Expired: drop it so it can never be re-adopted, and treat as no lease. + // saveLeases skips expired entries, so the ledger self-cleans on the + // next mutation; no extra write is needed here. delete(h.leases, identity) return nil } @@ -887,6 +901,7 @@ func NewContributeWSHub(logger *slog.Logger, server *Server) *ContributeWSHub { failedTasksFile: failedTasksFile, noPRStreaksFile: noPRStreaksFile, noWorkVerdictsFile: noWorkVerdictsPath(), + leasesFile: taskLeasesPath(), asyncActivitySave: asyncActivitySave, persistActivity: activityPersistenceEnabled, persistTaskLedgers: taskLedgerPersistenceEnabled, @@ -901,6 +916,7 @@ func NewContributeWSHub(logger *slog.Logger, server *Server) *ContributeWSHub { hub.loadFailedTasks() hub.loadNoPRStreaks() hub.loadNoWorkVerdicts() + hub.loadLeases() hub.loadActivity() go hub.cleanupLoop() return hub