diff --git a/.github/wiki/guides/ci/Scheduling-and-Cost.md b/.github/wiki/guides/ci/Scheduling-and-Cost.md index 6a98da723..e1d8446af 100644 --- a/.github/wiki/guides/ci/Scheduling-and-Cost.md +++ b/.github/wiki/guides/ci/Scheduling-and-Cost.md @@ -60,4 +60,4 @@ An allowance is the declared ceiling (`economics.ceilings.`) minus the - Not wired yet (debt D-1280): the dispatcher must bind the economics environment with `economics.Bind` (store, spend policy, clock, coordinator id) and copy the spend keys into the scheduler's policy with `economics.ApplyPolicy`. Until then the hooks are inert and nothing paid is placed. - A job the scheduler holds with `approval.pending` never reaches the admission step. The dispatcher records the whole pipeline's worst case with `economics.NotePending`, so `ApproveSpend` has a figure to bind to and one approval releases the run. Without it, the figure is the pipeline total at the moment a job is held, and each job that grows the pipeline past it is held again. -- Burst (P7-SCHED-08) calls `economics.ReserveResource` and `economics.ReleaseResource` inside its own transactions. The GitHub billing importer (P7-HOST-10) writes through `economics.RecordUsage`. +- Burst (P7-SCHED-08) calls `economics.ReserveResource` and `economics.ReleaseResource` inside its own transactions. The GitHub billing importer (P7-HOST-10) writes through `economics.ReplaceUsage`. diff --git a/free/ci/cmd/providers_usage.go b/free/ci/cmd/providers_usage.go index 2cd4d6bc9..d3312147d 100644 --- a/free/ci/cmd/providers_usage.go +++ b/free/ci/cmd/providers_usage.go @@ -8,7 +8,7 @@ package main // Outputs: `disabled` when providers.github.billing_import is not true (no API call, exit 0); else a // one-line summary or one v1 JSON envelope (an error too, with --json); exit 0, 1 // (usage, E700, E709 and other provider refusals) or 2 (rate limit, network, anything else). -// Constraints: V15 only; writes the store only through economics.RecordUsage; requests only the +// Constraints: V15 only; writes the store only through economics.ReplaceUsage; requests only the // non-summary billing usage endpoint (see internal/providers/github/usage); never asks for // or refreshes a token scope. Live use waits for the owner to name the token scope. diff --git a/free/ci/cmd/serve_rereport.go b/free/ci/cmd/serve_rereport.go new file mode 100644 index 000000000..d942fd5de --- /dev/null +++ b/free/ci/cmd/serve_rereport.go @@ -0,0 +1,64 @@ +package main + +// Purpose: the coordinator's Rereport hook (D-1320): a sweep found a commit without a terminal nself-ci +// status although a run of it finished within the redelivery horizon; the stored result is sent +// again instead of building the commit a second time. +// Inputs: the finished intake row and the evidence the run recorded. +// Outputs: one status through the outbox (a new seq, so it replaces any parked older one). +// Constraints: only a bound, whole-gate GitHub run is re-sent (a partial run, a cancelled one or an +// unbound one never posted a status on purpose); the state is the one the run was authorized to +// post (the outbox payload: a failed release publication vetoes a passing result there, not in +// the evidence), and a run without that record is not re-sent; the description names the run, +// never evidence text the publication profile might filter. + +import ( + "context" + "encoding/json" + "fmt" + "strings" + + "github.com/nself-org/plugins/free/ci/internal/model" + "github.com/nself-org/plugins/free/ci/internal/report" + "github.com/nself-org/plugins/free/ci/internal/store" +) + +func rereportDone(st *store.Store) func(context.Context, store.IntakeRow) error { + return func(ctx context.Context, row store.IntakeRow) error { + owner, repo, ok := strings.Cut(row.Project, "/") + if !ok || row.Source != "github" || row.Pipeline != "" || row.PipelineID == "" { + return nil + } + body, found, err := st.RunEvidence(ctx, row.PipelineID) + if err != nil || !found { + return err + } + var ev model.Evidence + if err = json.Unmarshal(body, &ev); err != nil || ev.Binding.State != "bound" || ev.Revision != row.Revision { + return nil + } + if ev.Result != "pass" && ev.Result != "fail" && ev.Result != "error" { + return nil + } + key := report.KeyFor(owner, repo, row.Revision) + payload, found, err := st.OutboxDecision(ctx, key.Revision, key.Context) + if err != nil || !found { + return err + } + var posted report.Status + if err = json.Unmarshal(payload, &posted); err != nil { + return nil + } + state := posted.State + if state != report.Success && state != report.Failure && state != report.Error { + return nil + } + if state == report.Success && ev.Result != "pass" { // evidence never allows a success the run could not post + return nil + } + sender := report.Default + if sender == nil || sender.Store != st { + sender = &report.Sender{Store: st} + } + return sender.Repost(ctx, []report.Key{key}, state, fmt.Sprintf("result of run %s sent again", row.PipelineID)) + } +} diff --git a/free/ci/cmd/serve_rereport_test.go b/free/ci/cmd/serve_rereport_test.go new file mode 100644 index 000000000..fc8ea7fad --- /dev/null +++ b/free/ci/cmd/serve_rereport_test.go @@ -0,0 +1,92 @@ +package main + +import ( + "context" + "encoding/json" + "path/filepath" + "testing" + + "github.com/nself-org/plugins/free/ci/internal/model" + "github.com/nself-org/plugins/free/ci/internal/report" + "github.com/nself-org/plugins/free/ci/internal/store" +) + +// D-1320: the status a finished, bound run was authorized to post is queued again, not the evidence result +// (a failed release publication vetoes a passing run in the outbox only); anything the run would not have +// posted is not. +func TestRereportDoneQueuesTheStoredResult(t *testing.T) { + ctx := context.Background() + rev := "0123456789abcdef0123456789abcdef01234567" + for _, c := range []struct { + name string + bind model.BindingState + result model.Result + pipeline string + posted report.State // the status the run queued; "" = none was recorded + want report.State + }{ + {"pass", "bound", "pass", "", report.Success, report.Success}, + {"fail", "bound", "fail", "", report.Failure, report.Failure}, + {"error", "bound", "error", "", report.Error, report.Error}, + {"publication failed after a passing gate", "bound", "pass", "", report.Failure, report.Failure}, + {"pass without a recorded decision", "bound", "pass", "", "", ""}, + {"success recorded for a failed run", "bound", "fail", "", report.Success, ""}, + {"unbound run", "unbound", "pass", "", report.Success, ""}, + {"partial run", "bound", "pass", "release", report.Success, ""}, + {"other result", "bound", "cancelled", "", report.Failure, ""}, + } { + t.Run(c.name, func(t *testing.T) { + st, err := store.Open(filepath.Join(t.TempDir(), "state.db"), store.Options{}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = st.Close() }) + if err = st.CreatePipeline(ctx, store.PipelineRow{ID: "run-1", Revision: rev, Project: "/proj", Trigger: "push", SourceTrust: "trusted", PrivacyZone: "local-only", + PolicyDigest: "p", InputDigest: "cfg", SelectionMode: "full", Status: "passed"}); err != nil { + t.Fatal(err) + } + ev := model.Evidence{Revision: rev, Binding: model.Binding{State: c.bind}, Result: c.result} + body, _ := json.Marshal(ev) + if err = st.RecordRunEvidence(ctx, "ev1", "run-1", rev, "cfg", body); err != nil { + t.Fatal(err) + } + if c.posted != "" { // what finishing the run queued through the outbox, after the release veto + seq, err := st.OutboxReserve(ctx, rev, "nself-ci@o/r") + if err != nil { + t.Fatal(err) + } + payload, _ := json.Marshal(report.Status{Owner: "o", Repo: "r", Revision: rev, State: c.posted, Description: "run"}) + if err = st.OutboxReady(ctx, rev, "nself-ci@o/r", seq, payload); err != nil { + t.Fatal(err) + } + } + row := store.IntakeRow{Source: "github", Project: "o/r", Revision: rev, PipelineID: "run-1", Pipeline: c.pipeline} + if err = rereportDone(st)(ctx, row); err != nil { + t.Logf("delivery error (no remote in the test): %v", err) + } + all, err := st.OutboxList(ctx) + if err != nil { + t.Fatal(err) + } + var items []store.OutboxItem // without the entry the run itself recorded + for _, it := range all { + if it.Seq > 1 || c.posted == "" { + items = append(items, it) + } + } + if c.want == "" { + if len(items) != 0 { + t.Fatalf("queued %d status(es), want none", len(items)) + } + return + } + if len(items) != 1 || items[0].Revision != rev || items[0].Context != "nself-ci@o/r" { + t.Fatalf("outbox = %+v", items) + } + var s report.Status + if err = json.Unmarshal(items[0].Payload, &s); err != nil || s.State != c.want || s.Owner != "o" || s.Repo != "r" { + t.Fatalf("payload %s (%v), want state %s for o/r", items[0].Payload, err, c.want) + } + }) + } +} diff --git a/free/ci/cmd/serve_wire.go b/free/ci/cmd/serve_wire.go index af1850959..6b3632c5f 100644 --- a/free/ci/cmd/serve_wire.go +++ b/free/ci/cmd/serve_wire.go @@ -86,6 +86,7 @@ func serveWire(ctx context.Context, st *store.Store, home, coordinatorID string, } undoHosted := serveHosted(st, home, cfg.AllowedRepos) undoNetscope := serveNetscope(st, home) + cfg.Rereport = rereportDone(st) cfg.Drain = func(dctx context.Context) error { d, err := drain.New(drain.Config{ Intake: agentIntake{}, Canceller: liveCanceller{}, Deadline: drain.FlagDeadline(), Logf: cfg.Logf, diff --git a/free/ci/internal/coordinator/clock.go b/free/ci/internal/coordinator/clock.go new file mode 100644 index 000000000..d41a676a5 --- /dev/null +++ b/free/ci/internal/coordinator/clock.go @@ -0,0 +1,18 @@ +package coordinator + +// Purpose: the time source of the coordinator's schedules and maintenance (moved out of server.go to keep it +// inside the file-size budget). +// Inputs: none. Outputs: Clock and its real implementation. Constraints: tests inject a fake Clock. + +import "time" + +// Clock is the time source of the schedules; tests inject a fake. +type Clock interface { + Now() time.Time + After(d time.Duration) <-chan time.Time +} + +type realClock struct{} + +func (realClock) Now() time.Time { return time.Now() } +func (realClock) After(d time.Duration) <-chan time.Time { return time.After(d) } diff --git a/free/ci/internal/coordinator/dedup.go b/free/ci/internal/coordinator/dedup.go index ab7d52dcf..c3157cd8b 100644 --- a/free/ci/internal/coordinator/dedup.go +++ b/free/ci/internal/coordinator/dedup.go @@ -1,6 +1,6 @@ package coordinator -// Purpose: the semantic key, the maintenance loop (retention, claim heartbeat, recovery) of the V15 +// Purpose: the semantic key, the maintenance loop (retention, claim heartbeat, recovery, approval voids) of the V15 // coordinator (P7-NODE-10, Epic D7, X7). // Inputs: project, revision, trigger class and policy digest of a trigger; the injected clock. // Outputs: a stable semantic key; a store that forgets what is older than the 72 h redelivery horizon. @@ -16,6 +16,7 @@ import ( "time" "github.com/nself-org/plugins/free/ci/internal/store" + "github.com/nself-org/plugins/free/ci/internal/trust/source" ) // semanticKey is the key that coalesces triggers for the same code under the same policy. A trigger that @@ -45,10 +46,9 @@ func (c *Coordinator) maintain(ctx context.Context) { if err := st.IntakeBeat(ctx, c.cfg.ID); err != nil && ctx.Err() == nil { c.cfg.Logf("intake heartbeat failed: %v", err) } + c.sweepApprovals(ctx) if now := c.cfg.Clock.Now(); now.Sub(pruned) >= time.Hour { - if _, err := st.DedupPrune(ctx, 0); err != nil && ctx.Err() == nil { - c.cfg.Logf("intake prune failed: %v", err) - } + c.prune(ctx, now) pruned = now } c.nudge() @@ -60,6 +60,53 @@ func (c *Coordinator) maintain(ctx context.Context) { } } +// sweepApprovals writes the void (and its audit row) of every standing approval that no longer matches its +// run, so the void does not wait for `ci runs pending`, a facts write or a new approval (D-1342). +func (c *Coordinator) sweepApprovals(ctx context.Context) { + if _, err := c.cfg.Store.SweepApprovals(ctx, "", approvalVoidCheck); err != nil && ctx.Err() == nil { + c.cfg.Logf("approval sweep failed: %v", err) + } +} + +// approvalVoidCheck adapts the pure void rule of trust/source to the store's binding type. +func approvalVoidCheck(approved, cur store.ApprovalBinding) []string { + return source.VoidReasons(source.Binding(approved), source.Binding(cur)) +} + +// MarkRetention is how long the marks of one provider_mark name stay. KeyPrefix narrows the rule to the +// keys that start with it ("" is every key of the name); a mark older than MaxAge is deleted. +type MarkRetention struct { + Name string + KeyPrefix string + MaxAge time.Duration +} + +// DefaultMarkRetention is the horizon of the GitHub reconcile marks (keys are "||"): a run +// update and a demand job are read again only while GitHub still lists them, so a week is generous; the +// head (term, enq) and redelivery marks keep 30 days, far beyond the 72 h redelivery horizon. +var DefaultMarkRetention = []MarkRetention{ + {Name: "github-reconcile", KeyPrefix: "run|", MaxAge: 7 * 24 * time.Hour}, + {Name: "github-reconcile", KeyPrefix: "demand|", MaxAge: 7 * 24 * time.Hour}, + {Name: "github-reconcile", MaxAge: 30 * 24 * time.Hour}, +} + +// prune forgets what is past its retention: finished intake rows and keys, and the provider marks. +func (c *Coordinator) prune(ctx context.Context, now time.Time) { + st := c.cfg.Store + if _, err := st.DedupPrune(ctx, 0); err != nil && ctx.Err() == nil { + c.cfg.Logf("intake prune failed: %v", err) + } + rules := c.cfg.MarkRetention + if rules == nil { + rules = DefaultMarkRetention + } + for _, r := range rules { + if _, err := st.ProviderMarkPrune(ctx, r.Name, r.KeyPrefix, now.Add(-r.MaxAge)); err != nil && ctx.Err() == nil { + c.cfg.Logf("provider mark prune failed (%s): %v", r.Name, err) + } + } +} + // dispatchLoop hands queued rows to the Dispatcher, at most Concurrency at a time, until ctx ends. func (c *Coordinator) dispatchLoop(ctx context.Context) { if c.cfg.Dispatch == nil { diff --git a/free/ci/internal/coordinator/ingress.go b/free/ci/internal/coordinator/ingress.go index 55a4c50ba..e62284022 100644 --- a/free/ci/internal/coordinator/ingress.go +++ b/free/ci/internal/coordinator/ingress.go @@ -29,6 +29,10 @@ type Delivery struct { Facts json.RawMessage // source facts (event, ref, clone_url, ...); opaque to the intake Pipeline string // the requested pipeline, "" for the default Synthetic *model.Pipeline // a pipeline the source built itself (runner demand), if any + // Reconcile marks a delivery found by a sweep (a head without a terminal status). It coalesces into a + // run of the same source and semantic key that finished within the redelivery horizon, and the + // coordinator then asks Config.Rereport to send that run's stored result again instead of building twice. + Reconcile bool } // Ingress turns the requests of one provider into Deliveries. diff --git a/free/ci/internal/coordinator/intake.go b/free/ci/internal/coordinator/intake.go index d5f311e4d..42646dd4d 100644 --- a/free/ci/internal/coordinator/intake.go +++ b/free/ci/internal/coordinator/intake.go @@ -81,7 +81,7 @@ func (c *Coordinator) enqueue(ctx context.Context, d Delivery, webhook bool, o e out, err := c.cfg.Store.IntakeEnqueue(ctx, store.IntakeRequest{ Source: d.Source, Event: event, DeliveryID: d.DeliveryID, Project: d.Project, Revision: d.Revision, TriggerClass: d.TriggerClass, PolicyDigest: c.cfg.PolicyDigest, SemanticKey: semanticKey(strings.ToLower(d.Project), d.Revision, d.TriggerClass, c.cfg.PolicyDigest, keyPipeline), - Pipeline: d.Pipeline, Facts: blob, Webhook: webhook, Force: c.cfg.Force || o.force, Actor: o.actor, + Pipeline: d.Pipeline, Facts: blob, Webhook: webhook, Reconcile: d.Reconcile, Force: c.cfg.Force || o.force, Actor: o.actor, IdemDigest: o.idem, ReqDigest: o.reqDigest, }) if err != nil { @@ -94,9 +94,33 @@ func (c *Coordinator) enqueue(ctx context.Context, d Delivery, webhook bool, o e if out.Kind == store.IntakeEnqueued { c.nudge() } + if d.Reconcile && (out.Kind == store.IntakeCoalesced || out.Kind == store.IntakeDuplicate) { // a duplicate: this sweep delivery id was folded or built before + + c.rereport(ctx, out.ID) + } return out, nil } +// rereport sends the stored result of a finished intake row again, for a sweep that found its commit without +// a terminal status (D-1320). A row that is still queued or running reports for itself; a failure is logged, +// never returned: the sweep offers the head again. +func (c *Coordinator) rereport(ctx context.Context, id string) { + if c.cfg.Rereport == nil { + return + } + row, err := c.cfg.Store.IntakeGet(ctx, id) + if err != nil { + c.cfg.Logf("intake row %s not read for a re-report: %v", id, err) + return + } + if row.State != store.IntakeDone { + return + } + if err = c.cfg.Rereport(ctx, row); err != nil { + c.cfg.Logf("re-report of intake row %s failed: %v", id, err) + } +} + func writeJSON(w http.ResponseWriter, code int, v any) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(code) diff --git a/free/ci/internal/coordinator/marks_prune_test.go b/free/ci/internal/coordinator/marks_prune_test.go new file mode 100644 index 000000000..33be4f2d1 --- /dev/null +++ b/free/ci/internal/coordinator/marks_prune_test.go @@ -0,0 +1,112 @@ +package coordinator + +// Purpose: the maintenance prune of provider marks (D-1319). + +import ( + "context" + "database/sql" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/nself-org/plugins/free/ci/internal/store" +) + +func TestPruneForgetsOldReconcileMarks(t *testing.T) { + ctx, r := context.Background(), newRig(t, nil) + now := r.clk.Now() + day := 24 * time.Hour + put := func(name, key string, age time.Duration) { + t.Helper() + if err := r.st.ProviderMarkPut(ctx, name, key, now.Add(-age)); err != nil { + t.Fatal(err) + } + } + put("github-reconcile", "run|o/r|1", 8*day) + put("github-reconcile", "run|o/r|2", 6*day) + put("github-reconcile", "demand|o/r|1", 8*day) + put("github-reconcile", "term|o/r|abc", 29*day) + put("github-reconcile", "enq|o/r|abc", 31*day) + put("github-reconcile", "redeliver|o/r|g", 31*day) + put("github-usage", "acct", 90*day) // another name: never touched by the github-reconcile rules + r.c.prune(ctx, now) + for _, m := range []struct { + name, key string + want bool + }{ + {"github-reconcile", "run|o/r|1", false}, {"github-reconcile", "run|o/r|2", true}, + {"github-reconcile", "demand|o/r|1", false}, {"github-reconcile", "term|o/r|abc", true}, + {"github-reconcile", "enq|o/r|abc", false}, {"github-reconcile", "redeliver|o/r|g", false}, + {"github-usage", "acct", true}, + } { + if _, ok, err := r.st.ProviderMarkGet(ctx, m.name, m.key); err != nil || ok != m.want { + t.Fatalf("mark %s/%s present = %v (%v), want %v", m.name, m.key, ok, err, m.want) + } + } +} + +func TestPruneKeepsMarksWhenRetentionIsEmpty(t *testing.T) { + ctx, r := context.Background(), newRig(t, func(c *Config) { c.MarkRetention = []MarkRetention{} }) + if err := r.st.ProviderMarkPut(ctx, "github-reconcile", "enq|o/r|a", r.clk.Now().Add(-90*24*time.Hour)); err != nil { + t.Fatal(err) + } + r.c.prune(ctx, r.clk.Now()) + if _, ok, _ := r.st.ProviderMarkGet(ctx, "github-reconcile", "enq|o/r|a"); !ok { + t.Fatal("an empty retention list must keep every mark") + } +} + +// The maintenance tick writes the void of a stale approval without anyone running `ci runs pending` +// (D-1342). +func TestMaintainSweepsStaleApprovals(t *testing.T) { + ctx, path := context.Background(), filepath.Join(t.TempDir(), "state.db") + r := newRig(t, func(c *Config) { + st, err := store.Open(path, store.Options{Clock: c.Clock.Now}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = st.Close() }) + c.Store = st + }) + p := store.PipelineRow{ID: "r1", Revision: "sha-1", Project: "/proj", Trigger: "push", SourceTrust: "untrusted", PrivacyZone: "local-only", + PolicyDigest: "p", InputDigest: "cfg-1", SelectionMode: "full", Status: "queued"} + if err := r.st.CreatePipeline(ctx, p); err != nil { + t.Fatal(err) + } + if _, err := r.st.RecordApprovalFacts(ctx, store.ApprovalRunFacts{Run: "r1", Author: "mallory", HeadRepo: "fork/app", RequiresApproval: true}, approvalVoidCheck); err != nil { + t.Fatal(err) + } + if _, err := r.st.ApproveRun(ctx, store.ApproveInput{Run: "r1", Approver: "bob", ExpectSHA: "sha-1", Check: approvalVoidCheck}); err != nil { + t.Fatal(err) + } + mctx, stop := context.WithCancel(ctx) + done := make(chan struct{}) + go func() { r.c.maintain(mctx); close(done) }() + defer func() { + stop() + <-done + }() + waitFor(t, "the first beat to finish", func() bool { return r.clk.pending() >= 1 }) + if st, _, _ := r.st.ApprovalState(ctx, "r1"); st.Approval.VoidedAt != 0 { + t.Fatal("an approval that still matches its run must stand") + } + // The pipeline moves to a new head, as a force-push would. + db, err := sql.Open("sqlite", "file:"+path+"?_pragma=busy_timeout(5000)") + if err != nil { + t.Fatal(err) + } + defer func() { _ = db.Close() }() + if _, err = db.Exec("UPDATE pipeline SET revision='sha-2' WHERE id='r1'"); err != nil { + t.Fatal(err) + } + r.clk.Advance(r.c.cfg.Beat) + waitFor(t, "the beat to void the stale approval", func() bool { + st, _, _ := r.st.ApprovalState(ctx, "r1") + return st.Approval.VoidedAt != 0 + }) + st, _, err := r.st.ApprovalState(ctx, "r1") + if err != nil || !strings.Contains(st.Approval.VoidReason, "head") { + t.Fatalf("the tick did not void the stale approval: %+v %v", st.Approval, err) + } +} diff --git a/free/ci/internal/coordinator/rereport_test.go b/free/ci/internal/coordinator/rereport_test.go new file mode 100644 index 000000000..756e3378e --- /dev/null +++ b/free/ci/internal/coordinator/rereport_test.go @@ -0,0 +1,95 @@ +package coordinator + +// Purpose: a head-scan delivery that folds into a finished run sends the stored result again (D-1320). + +import ( + "context" + "testing" + "time" + + "github.com/nself-org/plugins/free/ci/internal/store" +) + +// recordBoundRun leaves a pipeline row and the bound, passing evidence of a whole-gate run of testSHA. +func recordBoundRun(t *testing.T, st *store.Store, id string) { + t.Helper() + ctx := context.Background() + if err := st.CreatePipeline(ctx, store.PipelineRow{ID: id, Revision: testSHA, Project: "/proj", Trigger: "push", SourceTrust: "trusted", PrivacyZone: "local-only", + PolicyDigest: "p", InputDigest: "cfg", SelectionMode: "full", Status: "passed"}); err != nil { + t.Fatal(err) + } + body := []byte(`{"revision":"` + testSHA + `","binding":{"state":"bound"},"result":"pass"}`) + if err := st.RecordRunEvidence(ctx, "ev-"+id, id, testSHA, "cfg", body); err != nil { + t.Fatal(err) + } + if _, err := st.OutboxEnqueue(ctx, testSHA, "nself-ci@o/r", []byte(`{"state":"success"}`)); err != nil { // the status the run was authorized to post + t.Fatal(err) + } +} + +func TestReconcileDeliveryRereportsFinishedRun(t *testing.T) { + ctx := context.Background() + var got []store.IntakeRow + r := newRig(t, func(c *Config) { + c.Rereport = func(_ context.Context, row store.IntakeRow) error { got = append(got, row); return nil } + }) + r.post("/webhook", "push", "g1", pushBody(testSHA)) + row, ok, err := r.st.IntakeClaim(ctx, "w") + if err != nil || !ok { + t.Fatal(err) + } + sweep := Delivery{Source: row.Source, Project: "o/r", Revision: testSHA, DeliveryID: "reconcile:o/r:" + testSHA, Reconcile: true} + // The webhook's row is still queued: it reports for itself, nothing is sent again. + if out, err := r.c.Enqueue(ctx, sweep); err != nil || out.Kind != store.IntakeCoalesced || len(got) != 0 { + t.Fatalf("while the run is live: %+v %v, %d re-reports", out, err, len(got)) + } + recordBoundRun(t, r.st, "run-1") + if err = r.st.IntakeFinish(ctx, row.ID, "w", "run-1", "exit=0", true); err != nil { + t.Fatal(err) + } + r.clk.Advance(2 * time.Hour) + sweep.DeliveryID = "reconcile:o/r:" + testSHA + ":2" + if out, err := r.c.Enqueue(ctx, sweep); err != nil || out.Kind != store.IntakeCoalesced || out.ID != row.ID { + t.Fatalf("after the run: %+v %v, want coalesced into %s (no second build)", out, err, row.ID) + } + if len(got) != 1 || got[0].ID != row.ID || got[0].PipelineID != "run-1" { + t.Fatalf("re-reports = %+v, want the finished row once", got) + } + if out, _ := r.c.Enqueue(ctx, sweep); out.Kind != store.IntakeDuplicate || len(got) != 2 { + t.Fatalf("the same sweep delivery again: %+v, %d re-reports; want a duplicate that sends the result again", out, len(got)) + } + if st, _ := r.st.IntakeStats(ctx); st.Queued != 0 { + t.Fatalf("a second build was queued: %+v", st) + } +} + +func TestPlainDeliveryNeverRereports(t *testing.T) { + ctx := context.Background() + called := false + r := newRig(t, func(c *Config) { + c.Rereport = func(context.Context, store.IntakeRow) error { called = true; return nil } + }) + r.post("/webhook", "push", "g1", pushBody(testSHA)) + row, _, _ := r.st.IntakeClaim(ctx, "w") + _ = r.st.IntakeFinish(ctx, row.ID, "w", "run-1", "exit=0", true) + if out, err := r.c.Enqueue(ctx, Delivery{Source: row.Source, Project: "o/r", Revision: testSHA, DeliveryID: "other"}); err != nil || out.Kind != store.IntakeEnqueued || called { + t.Fatalf("a plain delivery: %+v %v rereported=%v", out, err, called) + } +} + +// A finished run with no result to send again (cancelled here) is not folded into: the sweep builds the head. +func TestReconcileDeliveryBuildsWhenFinishedRunHasNothingToResend(t *testing.T) { + ctx := context.Background() + called := false + r := newRig(t, func(c *Config) { + c.Rereport = func(context.Context, store.IntakeRow) error { called = true; return nil } + }) + r.post("/webhook", "push", "g1", pushBody(testSHA)) + row, _, _ := r.st.IntakeClaim(ctx, "w") + _ = r.st.IntakeFinish(ctx, row.ID, "w", "run-1", "exit=0", true) // no evidence recorded + r.clk.Advance(2 * time.Hour) + sweep := Delivery{Source: row.Source, Project: "o/r", Revision: testSHA, DeliveryID: "reconcile:o/r:" + testSHA, Reconcile: true} + if out, err := r.c.Enqueue(ctx, sweep); err != nil || out.Kind != store.IntakeEnqueued || out.ID == row.ID || called { + t.Fatalf("sweep after an unreportable run: %+v %v rereported=%v; want a new run", out, err, called) + } +} diff --git a/free/ci/internal/coordinator/server.go b/free/ci/internal/coordinator/server.go index 0c67152bb..770489a05 100644 --- a/free/ci/internal/coordinator/server.go +++ b/free/ci/internal/coordinator/server.go @@ -34,17 +34,6 @@ type RunResult struct { // Dispatcher runs the pipeline an intake row stands for. An error is an infrastructure failure. type Dispatcher func(ctx context.Context, row store.IntakeRow) (RunResult, error) -// Clock is the time source of the schedules; tests inject a fake. -type Clock interface { - Now() time.Time - After(d time.Duration) <-chan time.Time -} - -type realClock struct{} - -func (realClock) Now() time.Time { return time.Now() } -func (realClock) After(d time.Duration) <-chan time.Time { return time.After(d) } - // Config is the coordinator configuration. type Config struct { Store *store.Store @@ -65,6 +54,9 @@ type Config struct { StaleAfter time.Duration ID string Logf func(format string, a ...any) + + Rereport func(ctx context.Context, row store.IntakeRow) error // re-sends a finished row's result when a sweep folds into it (D-1320) + MarkRetention []MarkRetention // replaces DefaultMarkRetention when non-nil; empty keeps every mark // Drain, when set, runs once after the context ends and before the dispatches are cancelled, with new // claims already stopped (PauseIntake); P7-NODE-20 wires the graceful drain through it. Drain func(ctx context.Context) error diff --git a/free/ci/internal/economics/usage.go b/free/ci/internal/economics/usage.go index 96ea78138..a3364ca73 100644 --- a/free/ci/internal/economics/usage.go +++ b/free/ci/internal/economics/usage.go @@ -5,7 +5,7 @@ package economics // Inputs: a Usage row from P7-HOST-10's importer; the store, the spend policy and a clock. // Outputs: provider usage rows (the only `known` figures); a Report in which // every money figure carries its confidence and source. -// Constraints: RecordUsage is idempotent by Key; the report never prints an +// Constraints: RecordUsage is idempotent by Key; ReplaceUsage swaps a provider's days in one transaction; the report never prints an // allowance, an estimate or an actual without a confidence label; rates, // ceilings and budget limits are the owner's declaration and are printed as // such; nothing here calls a billing or pricing API. @@ -29,26 +29,55 @@ type Usage struct { ReportedAt time.Time } -// RecordUsage stores a provider-reported line with confidence `known`. -func RecordUsage(ctx context.Context, u Usage) error { - env, ok := envOf(ctx) - if !ok { - return refuse(model.AllowanceUnknown, "economics is not bound, so usage cannot be recorded") - } +// usageRow validates a Usage and converts it to its ledger row. +func usageRow(env Env, u Usage) (store.UsageRow, error) { day, err := time.Parse("2006-01-02", u.Date) if u.Key == "" || u.Provider == "" || err != nil || u.Minutes < 0 || u.CostMicro < 0 || (u.BalanceMicro != nil && *u.BalanceMicro < 0) { - return refuse(model.AllowanceUnknown, "a usage line needs a key, a provider, a UTC date and non-negative amounts") + return store.UsageRow{}, refuse(model.AllowanceUnknown, "a usage line needs a key, a provider, a UTC date and non-negative amounts") } // A provider in UTC+N may already report tomorrow; anything later is not a usage date. if u.Date > dayOf(env.Now().AddDate(0, 0, 1)) { - return refuse(model.AllowanceUnknown, "usage dated %s is in the future", u.Date) + return store.UsageRow{}, refuse(model.AllowanceUnknown, "usage dated %s is in the future", u.Date) } at := u.ReportedAt if at.IsZero() || at.After(env.Now()) { // a report is never made in the future at = env.Now() } - return env.Store.PutUsage(ctx, store.UsageRow{Key: u.Key, Provider: u.Provider, Period: day.Format("2006-01"), Day: u.Date, Series: u.Series, Minutes: u.Minutes, - CostMicro: u.CostMicro, BalanceMicro: u.BalanceMicro, ReportedAt: at.UnixNano()}) + return store.UsageRow{Key: u.Key, Provider: u.Provider, Period: day.Format("2006-01"), Day: u.Date, Series: u.Series, Minutes: u.Minutes, + CostMicro: u.CostMicro, BalanceMicro: u.BalanceMicro, ReportedAt: at.UnixNano()}, nil +} + +// RecordUsage stores a provider-reported line with confidence `known`. +func RecordUsage(ctx context.Context, u Usage) error { + env, ok := envOf(ctx) + if !ok { + return refuse(model.AllowanceUnknown, "economics is not bound, so usage cannot be recorded") + } + row, err := usageRow(env, u) + if err != nil { + return err + } + return env.Store.PutUsage(ctx, row) +} + +// ReplaceUsage stores lines as the whole usage of provider for the UTC days fromDay..toDay (both +// included) in one transaction: rows of the provider in that range that are not among lines are +// deleted (D-1317). Every line must be valid, belong to provider and lie in the range, or nothing is +// written. +func ReplaceUsage(ctx context.Context, provider, fromDay, toDay string, lines []Usage) error { + env, ok := envOf(ctx) + if !ok { + return refuse(model.AllowanceUnknown, "economics is not bound, so usage cannot be recorded") + } + rows := make([]store.UsageRow, 0, len(lines)) + for _, u := range lines { + row, err := usageRow(env, u) + if err != nil { + return err + } + rows = append(rows, row) + } + return env.Store.ReplaceUsage(ctx, provider, fromDay, toDay, rows) } // BudgetLine is one budget scope. Reserved is the worst case held (a bound, so estimated). diff --git a/free/ci/internal/providers/github/reconcile/headscan.go b/free/ci/internal/providers/github/reconcile/headscan.go index e395aa579..3ca166ed5 100644 --- a/free/ci/internal/providers/github/reconcile/headscan.go +++ b/free/ci/internal/providers/github/reconcile/headscan.go @@ -217,6 +217,7 @@ func (e *Engine) sweepHead(ctx context.Context, s *session, enq coordinator.Enqu if err != nil { return false, err } + d.Reconcile = true // a finished run of this head absorbs the delivery and its stored result is sent again (D-1320) if _, err = enq.Enqueue(ctx, d); err != nil { return false, err } diff --git a/free/ci/internal/providers/github/reconcile/state_test.go b/free/ci/internal/providers/github/reconcile/state_test.go new file mode 100644 index 000000000..2274dd1ed --- /dev/null +++ b/free/ci/internal/providers/github/reconcile/state_test.go @@ -0,0 +1,17 @@ +package reconcile + +// Purpose: the mark name the coordinator prunes stays the one the Engine writes (D-1319). + +import ( + "testing" + + "github.com/nself-org/plugins/free/ci/internal/coordinator" +) + +func TestMarkNameIsPrunedByTheCoordinator(t *testing.T) { + for _, r := range coordinator.DefaultMarkRetention { + if r.Name != markName { + t.Fatalf("the coordinator prunes marks named %q but the Engine writes %q", r.Name, markName) + } + } +} diff --git a/free/ci/internal/providers/github/usage/helpers_test.go b/free/ci/internal/providers/github/usage/helpers_test.go index 7c645e4fa..ab9f3295f 100644 --- a/free/ci/internal/providers/github/usage/helpers_test.go +++ b/free/ci/internal/providers/github/usage/helpers_test.go @@ -128,11 +128,11 @@ type recorder struct { lines []economics.Usage } -func (r *recorder) record(ctx context.Context, u economics.Usage) error { +func (r *recorder) replace(ctx context.Context, provider, from, to string, lines []economics.Usage) error { r.mu.Lock() - r.lines = append(r.lines, u) + r.lines = append(r.lines, lines...) r.mu.Unlock() - return economics.RecordUsage(ctx, u) + return economics.ReplaceUsage(ctx, provider, from, to, lines) } func (r *recorder) keys() map[string]economics.Usage { diff --git a/free/ci/internal/providers/github/usage/importer.go b/free/ci/internal/providers/github/usage/importer.go index 8f13c7021..2b0bb80fb 100644 --- a/free/ci/internal/providers/github/usage/importer.go +++ b/free/ci/internal/providers/github/usage/importer.go @@ -2,20 +2,20 @@ package usage // Purpose: Importer, the off-by-default step that turns a UsageSource's lines into SCHED-07 ledger // rows (Epic D12). -// Inputs: Settings (billing_import, billing_account), a providers.UsageSource, a recorder (default -// economics.RecordUsage) and a clock. +// Inputs: Settings (billing_import, billing_account), a providers.UsageSource, a replacer (default +// economics.ReplaceUsage) and a clock. // Outputs: Result; one economics.Usage per (account, date, sku, repo) with the key -// github::::, so a second import of the same period replaces the -// same rows (RecordUsage is an upsert by key) and never adds one. +// github::::; the period is written as a whole, in one transaction +// (ReplaceUsage), so a second import replaces the same rows and a row GitHub no longer reports +// (renamed repo or sku, changed billing account) is deleted instead of counted twice. // Constraints: with billing_import false Run returns Disabled before the source is touched (no API // call, no credential read). The whole period is read and converted before the first write, -// so a failing or malformed report writes nothing; a ledger write that fails midway leaves -// the earlier lines in place, the error names the key, and a rerun converges (same keys). +// so a failing or malformed report writes nothing, and the ledger write is atomic: a failure +// leaves the earlier rows and a rerun converges. An empty report writes and deletes nothing. // No remaining value is computed or reported here. import ( "context" - "errors" "fmt" "math" "net/url" @@ -39,14 +39,15 @@ const ( LookbackDays = 7 ) -// Recorder writes one usage line; economics.RecordUsage is the production one. -type Recorder func(context.Context, economics.Usage) error +// Replacer writes the usage of provider for the UTC days from..to (both included) as exactly lines, +// dropping what else the ledger holds for those days; economics.ReplaceUsage is the production one. +type Replacer func(ctx context.Context, provider, from, to string, lines []economics.Usage) error // Importer imports one account's usage. type Importer struct { Settings Settings Source providers.UsageSource // may be nil only while Settings.Import is false - Record Recorder // default economics.RecordUsage + Replace Replacer // default economics.ReplaceUsage Now func() time.Time // default time.Now } @@ -138,19 +139,21 @@ func (i *Importer) Run(ctx context.Context) (Result, error) { if err != nil { return res, err } - record := i.Record - if record == nil { - record = economics.RecordUsage + // An empty report replaces nothing: a malformed or truncated answer must not erase known usage (the + // allowance would read larger), so a billing account that really went quiet keeps its old rows + // until they leave the month. + if len(lines) == 0 { + return res, nil } - var errs []error - for _, l := range lines { - if err := record(ctx, l); err != nil { - errs = append(errs, fmt.Errorf("record %s: %w", l.Key, err)) - continue - } - res.Rows++ + replace := i.Replace + if replace == nil { + replace = economics.ReplaceUsage + } + if err := replace(ctx, Provider, res.From, res.To, lines); err != nil { + return res, fmt.Errorf("replace usage %s..%s: %w", res.From, res.To, err) } - return res, errors.Join(errs...) + res.Rows = len(lines) + return res, nil } // Key is the idempotent ledger key of one line. diff --git a/free/ci/internal/providers/github/usage/keys.go b/free/ci/internal/providers/github/usage/keys.go index bc9c88a4f..3d487291e 100644 --- a/free/ci/internal/providers/github/usage/keys.go +++ b/free/ci/internal/providers/github/usage/keys.go @@ -1,6 +1,6 @@ // Package usage is the GitHub enhanced-billing importer (P7-HOST-10, Epic D12). It reads the // consumption GitHub reports for one account and writes the Actions minute lines into the SCHED-07 -// usage ledger through economics.RecordUsage, so `nself ci policy usage` can show them as `known`. +// usage ledger through economics.ReplaceUsage, so `nself ci policy usage` can show them as `known`. // It never reports a remaining value and never keeps a ledger of its own. package usage diff --git a/free/ci/internal/providers/github/usage/replace_test.go b/free/ci/internal/providers/github/usage/replace_test.go new file mode 100644 index 000000000..ca35b57cf --- /dev/null +++ b/free/ci/internal/providers/github/usage/replace_test.go @@ -0,0 +1,76 @@ +package usage_test + +// Tests of the period replacement of the importer (D-1317): consumption GitHub reports under a new repo +// name, sku or billing account must not be counted next to the rows the old names left behind. + +import ( + "context" + "testing" + "time" + + "github.com/nself-org/plugins/free/ci/internal/providers" + "github.com/nself-org/plugins/free/ci/internal/providers/github/usage" +) + +// stubSource answers every request with its rows. +type stubSource struct{ rows []providers.UsageRow } + +func (s *stubSource) Usage(context.Context, providers.Period) ([]providers.UsageRow, error) { + return s.rows, nil +} + +func row(account, repo, sku string, micro int64) providers.UsageRow { + return providers.UsageRow{Account: account, Date: time.Date(2026, 10, 17, 0, 0, 0, 0, time.UTC), Product: "Actions", SKU: sku, + Quantity: 10, Unit: "minutes", NetMicro: micro, Repo: repo} +} + +func TestImportReplacesRowsUnderOldNames(t *testing.T) { + st := openLedger(t) + ctx := ledgerCtx(st) + run := func(account string, rows ...providers.UsageRow) { + t.Helper() + if _, err := importer(&stubSource{rows}, account, true, &recorder{}).Run(ctx); err != nil { + t.Fatal(err) + } + } + total := func() int64 { + t.Helper() + in, err := st.Allowance(context.Background(), usage.Provider, "2026-10") + if err != nil { + t.Fatal(err) + } + return in.Imported + } + run("org:acme", row("org:acme", "acme/old", "Actions Linux", 500_000)) + if got := total(); got != 500_000 { + t.Fatalf("first import counts %d, want 500000", got) + } + run("org:acme", row("org:acme", "acme/new", "Actions Linux", 500_000)) // the repo was renamed + if got := total(); got != 500_000 { + t.Fatalf("after a repo rename the allowance counts %d, want the same 500000 once", got) + } + run("org:acme", row("org:acme", "acme/new", "Actions Linux 2-core", 500_000)) // the sku was renamed + if got := total(); got != 500_000 { + t.Fatalf("after a sku rename the allowance counts %d, want 500000 once", got) + } + run("org:other", row("org:other", "acme/new", "Actions Linux 2-core", 500_000)) // billing_account changed + if got := total(); got != 500_000 { + t.Fatalf("after a billing account change the allowance counts %d, want 500000 once", got) + } +} + +// An empty report is not a statement that nothing was used: it deletes nothing. +func TestEmptyReportKeepsKnownUsage(t *testing.T) { + st := openLedger(t) + ctx := ledgerCtx(st) + if _, err := importer(&stubSource{[]providers.UsageRow{row("org:acme", "acme/r", "Actions Linux", 700_000)}}, "org:acme", true, &recorder{}).Run(ctx); err != nil { + t.Fatal(err) + } + res, err := importer(&stubSource{}, "org:acme", true, &recorder{}).Run(ctx) + if err != nil || res.Rows != 0 { + t.Fatalf("empty report: %+v %v", res, err) + } + if in, _ := st.Allowance(context.Background(), usage.Provider, "2026-10"); in.Imported != 700_000 { + t.Fatalf("an empty report changed the imported figure to %d", in.Imported) + } +} diff --git a/free/ci/internal/providers/github/usage/usage_test.go b/free/ci/internal/providers/github/usage/usage_test.go index 642119577..9825255e1 100644 --- a/free/ci/internal/providers/github/usage/usage_test.go +++ b/free/ci/internal/providers/github/usage/usage_test.go @@ -22,7 +22,7 @@ const ( ) func importer(src providers.UsageSource, account string, on bool, rec *recorder) *usage.Importer { - return &usage.Importer{Settings: usage.Settings{Import: on, Account: account}, Source: src, Record: rec.record, Now: func() time.Time { return today }} + return &usage.Importer{Settings: usage.Settings{Import: on, Account: account}, Source: src, Replace: rec.replace, Now: func() time.Time { return today }} } // TestDisabled proves acceptance 1: with billing_import false the command prints `disabled` and exits diff --git a/free/ci/internal/store/host_provider.go b/free/ci/internal/store/host_provider.go index f66cac58a..7c8c46745 100644 --- a/free/ci/internal/store/host_provider.go +++ b/free/ci/internal/store/host_provider.go @@ -136,6 +136,26 @@ func (s *Store) ProviderMarkGet(ctx context.Context, name, key string) (time.Tim return time.Unix(0, at).UTC(), true, nil } +// ProviderMarkPrune deletes the marks of name whose time is before olderThan and whose key starts with +// keyPrefix ("" matches every key) and returns how many it removed. Without it the table grows for the +// life of the state database; the coordinator calls it from its maintenance loop. +func (s *Store) ProviderMarkPrune(ctx context.Context, name, keyPrefix string, olderThan time.Time) (int64, error) { + if name == "" { + return 0, coded("E607", "provider mark prune needs a name") + } + var n int64 + err := s.write(ctx, func(tx *sql.Tx) error { + res, err := tx.ExecContext(ctx, `DELETE FROM provider_mark WHERE name=? AND at'' AND EXISTS ( + SELECT 1 FROM evidence e WHERE e.id=(SELECT e2.id FROM evidence e2 WHERE e2.pipeline_id=i.pipeline_id ORDER BY e2.created_at DESC LIMIT 1) + AND json_extract(CAST(e.body AS TEXT),'$.binding.state')='bound' + AND json_extract(CAST(e.body AS TEXT),'$.revision')=i.revision + AND json_extract(CAST(e.body AS TEXT),'$.result') IN ('pass','fail','error')) + AND EXISTS (SELECT 1 FROM outbox o WHERE o.revision=i.revision AND o.context='nself-ci@'||i.project AND o.payload_json IS NOT NULL))` + // IntakeEnqueue stores a trigger in one transaction with its delivery and key rows (and the audit row of // a forced enqueue). The caller may answer 2xx once it returns without error. func (s *Store) IntakeEnqueue(ctx context.Context, r IntakeRequest) (IntakeOutcome, error) { @@ -125,8 +139,8 @@ func (s *Store) IntakeEnqueue(ctx context.Context, r IntakeRequest) (IntakeOutco horizon = IntakeRetention } var into string - e := tx.QueryRowContext(ctx, `SELECT id FROM intake WHERE semantic_key=? AND (state IN ('queued','dispatched') OR (?=1 AND state='done' AND finished_at>? AND source=? AND event=?)) ORDER BY seq DESC LIMIT 1`, - r.SemanticKey, r.Webhook, now-int64(horizon), r.Source, r.Event).Scan(&into) + e := tx.QueryRowContext(ctx, `SELECT i.id FROM intake i WHERE i.semantic_key=? AND (i.state IN ('queued','dispatched') OR (i.state='done' AND i.finished_at>? AND i.source=? AND ((?=1 AND i.event=?) OR (?=1 AND `+resendable+`)))) ORDER BY i.seq DESC LIMIT 1`, + r.SemanticKey, now-int64(horizon), r.Source, r.Webhook, r.Event, r.Reconcile).Scan(&into) if e != nil && !errors.Is(e, sql.ErrNoRows) { return e } diff --git a/free/ci/internal/store/node_intake_reconcile_test.go b/free/ci/internal/store/node_intake_reconcile_test.go new file mode 100644 index 000000000..9c6a38f4a --- /dev/null +++ b/free/ci/internal/store/node_intake_reconcile_test.go @@ -0,0 +1,152 @@ +package store + +import ( + "context" + "fmt" + "testing" + "time" +) + +// recordRun stores a pipeline row and the evidence body of its run, as the dispatcher leaves them. +func recordRun(t *testing.T, s *Store, id, revision, result, binding string) { + t.Helper() + ctx := context.Background() + if err := s.CreatePipeline(ctx, PipelineRow{ID: id, Revision: revision, Project: "/proj", Trigger: "push", SourceTrust: "trusted", PrivacyZone: "local-only", + PolicyDigest: "p", InputDigest: "cfg", SelectionMode: "full", Status: "passed"}); err != nil { + t.Fatal(err) + } + body := fmt.Sprintf(`{"revision":%q,"binding":{"state":%q},"result":%q}`, revision, binding, result) + if err := s.RecordRunEvidence(ctx, "ev-"+id, id, revision, "cfg", []byte(body)); err != nil { + t.Fatal(err) + } +} + +// recordDecision leaves the status the run was authorized to post for the revision of req(): the record a +// sweep needs to send a result again (it carries a release veto the evidence lacks). +func recordDecision(t *testing.T, s *Store, revision string) { + t.Helper() + if _, err := s.OutboxEnqueue(context.Background(), revision, "nself-ci@o/r", []byte(`{"state":"success"}`)); err != nil { + t.Fatal(err) + } +} + +// doneRow leaves one finished github push run for key k, 71 h old. +func doneRow(t *testing.T) (*Store, string) { + t.Helper() + ctx := context.Background() + s, clk := intakeStore(t) + first := req("d1", "k") + first.Event = "push" + a, _ := s.IntakeEnqueue(ctx, first) + if _, _, err := s.IntakeClaim(ctx, "me"); err != nil { + t.Fatal(err) + } + recordRun(t, s, "run", "abc1234", "pass", "bound") + recordDecision(t, s, "abc1234") + if err := s.IntakeFinish(ctx, a.ID, "me", "run", "exit=0", true); err != nil { + t.Fatal(err) + } + clk.add(71 * time.Hour) + return s, a.ID +} + +// A head-scan (Reconcile) trigger folds into a run of the same source and key that finished inside the +// redelivery horizon; a plain non-webhook trigger, another source or another key does not (D-1320). +func TestIntakeReconcileFoldsOnlySameSourceAndKey(t *testing.T) { + for _, c := range []struct { + name, source, key string + reconcile bool + }{ + {"plain non-webhook trigger", "github", "k", false}, + {"another source", "gitlab", "k", true}, + {"another semantic key", "github", "other-key", true}, + } { + t.Run(c.name, func(t *testing.T) { + s, _ := doneRow(t) + r := req("s1", c.key) + r.Webhook, r.Reconcile, r.Source, r.Event = false, c.reconcile, c.source, "pull_request" + if b, err := s.IntakeEnqueue(context.Background(), r); err != nil || b.Kind != IntakeEnqueued { + t.Fatalf("%+v, %v; want a fresh run", b, err) + } + }) + } +} + +func TestIntakeReconcileFoldsWhatTheWebhookBuilt(t *testing.T) { + ctx := context.Background() + s, clk := intakeStore(t) + first := req("d1", "k") + first.Event = "push" + a, _ := s.IntakeEnqueue(ctx, first) + _, _, _ = s.IntakeClaim(ctx, "me") + recordRun(t, s, "run", "abc1234", "pass", "bound") + recordDecision(t, s, "abc1234") + _ = s.IntakeFinish(ctx, a.ID, "me", "run", "exit=0", true) + clk.add(71 * time.Hour) + r := req("reconcile:o/r:abc1234", "k") + r.Webhook, r.Reconcile, r.Event = false, true, "pull_request" + if b, err := s.IntakeEnqueue(ctx, r); err != nil || b.Kind != IntakeCoalesced || b.ID != a.ID { + t.Fatalf("reconcile 71 h after the run = %+v, %v; want coalesced into %s", b, err, a.ID) + } + if b, _ := s.IntakeEnqueue(ctx, r); b.Kind != IntakeDuplicate || b.ID != a.ID { + t.Fatalf("the same sweep delivery again = %+v, want duplicate of %s", b, a.ID) + } + clk.add(2 * time.Hour) + r.DeliveryID = "reconcile:o/r:abc1234:later" + if b, _ := s.IntakeEnqueue(ctx, r); b.Kind != IntakeEnqueued { + t.Fatalf("reconcile past the horizon = %+v, want a fresh run", b) + } +} + +// A run that failed to finish has no result to send again: the sweep builds the head. +func TestIntakeReconcileDoesNotFoldIntoFailedRow(t *testing.T) { + ctx := context.Background() + s, _ := intakeStore(t) + a, _ := s.IntakeEnqueue(ctx, req("d1", "k")) + _, _, _ = s.IntakeClaim(ctx, "me") + _ = s.IntakeFinish(ctx, a.ID, "me", "", "error: boom", false) + r := req("s1", "k") + r.Webhook, r.Reconcile = false, true + if b, _ := s.IntakeEnqueue(ctx, r); b.Kind != IntakeEnqueued { + t.Fatalf("reconcile after a failed row = %+v, want a fresh run", b) + } +} + +// A done row whose run posted no status on purpose (partial, unbound, cancelled, no evidence, another +// revision) has nothing to send again: the sweep builds the head instead of folding into it (D-1320). +func TestIntakeReconcileDoesNotFoldIntoRowWithoutResendableResult(t *testing.T) { + for _, c := range []struct { + name, result, binding, pipeline, evRevision string + noEvidence, noDecision bool + }{ + {"cancelled run", "cancelled", "bound", "", "abc1234", false, false}, + {"unbound run", "pass", "unbound", "", "abc1234", false, false}, + {"partial run", "pass", "bound", "release", "abc1234", false, false}, + {"run of another revision", "pass", "bound", "", "def5678", false, false}, + {"run without evidence", "", "", "", "", true, false}, + {"run without a recorded decision", "pass", "bound", "", "abc1234", false, true}, + } { + t.Run(c.name, func(t *testing.T) { + ctx := context.Background() + s, _ := intakeStore(t) + first := req("d1", "k") + first.Event, first.Pipeline = "push", c.pipeline + a, _ := s.IntakeEnqueue(ctx, first) + _, _, _ = s.IntakeClaim(ctx, "me") + if !c.noEvidence { + recordRun(t, s, "run", c.evRevision, c.result, c.binding) + if !c.noDecision { + recordDecision(t, s, "abc1234") + } + } + if err := s.IntakeFinish(ctx, a.ID, "me", "run", "exit=0", true); err != nil { + t.Fatal(err) + } + r := req("reconcile:o/r:abc1234", "k") + r.Webhook, r.Reconcile, r.Event = false, true, "pull_request" + if b, err := s.IntakeEnqueue(ctx, r); err != nil || b.Kind != IntakeEnqueued { + t.Fatalf("%+v, %v; want a fresh run, not a fold into %s", b, err, a.ID) + } + }) + } +} diff --git a/free/ci/internal/store/prune.go b/free/ci/internal/store/prune.go index 94a210b97..ad7cc444a 100644 --- a/free/ci/internal/store/prune.go +++ b/free/ci/internal/store/prune.go @@ -9,7 +9,7 @@ package store // attempts, leases and reservations, cancel intents, idempotency keys, // declared artifacts, cache entries and cache facts written by those attempts (an // entry is served only while its writer attempt is on record as passed), -// evidence and event-index rows of the pipeline, and one +// evidence, event-index, trust-fact and approval rows of the pipeline, and one // audit row records the prune; the audit log itself is never pruned. import ( @@ -61,6 +61,8 @@ var pruneSteps = []string{ `DELETE FROM event_index WHERE run_id IN (SELECT value FROM json_each(?))`, `DELETE FROM cache_entry WHERE attempt_id IN (SELECT a.id FROM attempt a JOIN job j ON a.job_id=j.id WHERE j.pipeline_id IN (SELECT value FROM json_each(?)))`, `DELETE FROM cache_attempt WHERE attempt_id IN (SELECT a.id FROM attempt a JOIN job j ON a.job_id=j.id WHERE j.pipeline_id IN (SELECT value FROM json_each(?)))`, + `DELETE FROM trust_approvals WHERE run IN (SELECT value FROM json_each(?))`, + `DELETE FROM trust_run_facts WHERE run IN (SELECT value FROM json_each(?))`, `DELETE FROM sched_admitted WHERE attempt_id IN (SELECT a.id FROM attempt a JOIN job j ON a.job_id=j.id WHERE j.pipeline_id IN (SELECT value FROM json_each(?)))`, `DELETE FROM sched_refusal WHERE attempt_id IN (SELECT a.id FROM attempt a JOIN job j ON a.job_id=j.id WHERE j.pipeline_id IN (SELECT value FROM json_each(?)))`, `DELETE FROM attempt WHERE job_id IN (SELECT id FROM job WHERE pipeline_id IN (SELECT value FROM json_each(?)))`, diff --git a/free/ci/internal/store/prune_test.go b/free/ci/internal/store/prune_test.go index ae2dad52b..3486f378c 100644 --- a/free/ci/internal/store/prune_test.go +++ b/free/ci/internal/store/prune_test.go @@ -90,3 +90,30 @@ func TestDeletePipelinesInBatches(t *testing.T) { t.Fatalf("attempt rows left: %d", got) } } + +// The trust rows of a pruned pipeline go with it, voided or not; another run's rows stay (D-1342). +func TestDeletePipelinesRemovesTrustRows(t *testing.T) { + ctx := context.Background() + s := testStore(t) + for _, run := range []string{"r1", "r2"} { + approvalSeed(t, s, run) + if _, err := s.ApproveRun(ctx, ApproveInput{Run: run, Approver: "bob", ExpectSHA: "sha-1", Check: exactCheck}); err != nil { + t.Fatal(err) + } + } + if err := s.FinishPipeline(ctx, "r1", "passed"); err != nil { + t.Fatal(err) + } + if n, err := s.DeletePipelines(ctx, []string{"r1"}); err != nil || n != 1 { + t.Fatalf("removed %d, err %v; want 1", n, err) + } + for _, table := range []string{"trust_run_facts", "trust_approvals"} { + if got := count(t, s, table); got != 1 { + t.Errorf("%s rows = %d, want only the unpruned run's", table, got) + } + var left string + if err := s.readers.QueryRow("SELECT run FROM " + table).Scan(&left); err != nil || left != "r2" { + t.Errorf("%s keeps %q (%v), want r2", table, left, err) + } + } +} diff --git a/free/ci/internal/store/queries.go b/free/ci/internal/store/queries.go index 5b17fa3e4..1e3189b87 100644 --- a/free/ci/internal/store/queries.go +++ b/free/ci/internal/store/queries.go @@ -3,6 +3,7 @@ package store import ( "context" "database/sql" + "errors" ) type AttemptView struct { @@ -108,6 +109,15 @@ func (s *Store) FlakyCandidates(ctx context.Context) ([]FlakyCandidate, error) { return out, rows.Err() } +// RunEvidence returns the newest evidence body recorded for a pipeline; ok is false when it has none. +func (s *Store) RunEvidence(ctx context.Context, pipelineID string) (body []byte, ok bool, err error) { + err = s.readers.QueryRowContext(ctx, `SELECT body FROM evidence WHERE pipeline_id=? ORDER BY created_at DESC LIMIT 1`, pipelineID).Scan(&body) + if errors.Is(err, sql.ErrNoRows) { + return nil, false, nil + } + return body, err == nil, err +} + func (s *Store) EvidenceByRevision(ctx context.Context, revision, inputDigest string) ([]EvidenceView, error) { rows, err := s.readers.QueryContext(ctx, `SELECT id,pipeline_id,coalesce(attempt_id,''),revision,input_digest,body,created_at FROM evidence WHERE revision=? AND input_digest=? ORDER BY created_at DESC`, revision, inputDigest) if err != nil { diff --git a/free/ci/internal/store/report_outbox.go b/free/ci/internal/store/report_outbox.go index 54c16fa06..cb4c99195 100644 --- a/free/ci/internal/store/report_outbox.go +++ b/free/ci/internal/store/report_outbox.go @@ -13,6 +13,7 @@ import ( "context" "database/sql" "embed" + "errors" ) //go:embed migrations/report-outbox/*.sql @@ -68,6 +69,17 @@ func (s *Store) OutboxReady(ctx context.Context, revision, statusContext string, }) } +// OutboxDecision is the payload of the newest entry of (revision, context) that holds one: the status the +// run was authorized to post, after the publisher, the release veto and the publish gates (D-1320). ok is +// false when no entry holds a payload (the run never reached its status). +func (s *Store) OutboxDecision(ctx context.Context, revision, statusContext string) (payload []byte, ok bool, err error) { + err = s.readers.QueryRowContext(ctx, "SELECT payload_json FROM outbox WHERE revision=? AND context=? AND payload_json IS NOT NULL ORDER BY seq DESC LIMIT 1", revision, statusContext).Scan(&payload) + if errors.Is(err, sql.ErrNoRows) { + return nil, false, nil + } + return payload, err == nil, err +} + // OutboxUnposted lists deliverable entries, oldest first. func (s *Store) OutboxUnposted(ctx context.Context) ([]OutboxEntry, error) { return collect(ctx, s, func(r *sql.Rows) (e OutboxEntry, err error) { diff --git a/free/ci/internal/store/sched_ledger_budget.go b/free/ci/internal/store/sched_ledger_budget.go index 558d9ae56..6c944db2e 100644 --- a/free/ci/internal/store/sched_ledger_budget.go +++ b/free/ci/internal/store/sched_ledger_budget.go @@ -136,21 +136,55 @@ type UsageRow struct { ReportedAt int64 } +const putUsageSQL = "INSERT OR REPLACE INTO ledger_usage(key,provider,period,day,series,minutes,cost_micro,balance_micro,reported_at) VALUES (?,?,?,?,?,?,?,?,?)" + +func putUsage(ctx context.Context, tx *sql.Tx, u UsageRow) error { + _, err := tx.ExecContext(ctx, putUsageSQL, u.Key, u.Provider, u.Period, u.Day, u.Series, u.Minutes, u.CostMicro, u.BalanceMicro, u.ReportedAt) + return err +} + // PutUsage stores or replaces a provider usage row. func (s *Store) PutUsage(ctx context.Context, u UsageRow) error { if err := ledgerValues(u.Minutes, u.CostMicro); err != nil { return err } return s.write(ctx, func(tx *sql.Tx) error { - _, err := tx.ExecContext(ctx, "INSERT OR REPLACE INTO ledger_usage(key,provider,period,day,series,minutes,cost_micro,balance_micro,reported_at) VALUES (?,?,?,?,?,?,?,?,?)", - u.Key, u.Provider, u.Period, u.Day, u.Series, u.Minutes, u.CostMicro, u.BalanceMicro, u.ReportedAt) - if err != nil { + if err := putUsage(ctx, tx, u); err != nil { return err } return ledgerTx(s, ctx, tx).pruneLedger() }) } +// ReplaceUsage replaces every row of provider whose day lies in fromDay..toDay (UTC, YYYY-MM-DD, both +// included) with rows, in one transaction: a row the provider no longer reports (a renamed repo or sku, +// a changed billing account) goes, so the allowance never counts one consumption twice, and a failure +// leaves the old rows. Every row must belong to provider and lie in the range. +func (s *Store) ReplaceUsage(ctx context.Context, provider, fromDay, toDay string, rows []UsageRow) error { + if provider == "" || fromDay == "" || fromDay > toDay { + return coded("E607", "usage replacement needs a provider and an ordered day range") + } + for _, u := range rows { + if u.Provider != provider || u.Day < fromDay || u.Day > toDay { + return coded("E607", "usage row "+u.Key+" is outside the replaced provider and day range") + } + } + return s.write(ctx, func(tx *sql.Tx) error { + if _, err := tx.ExecContext(ctx, "DELETE FROM ledger_usage WHERE provider=? AND day>=? AND day<=?", provider, fromDay, toDay); err != nil { + return err + } + for _, u := range rows { + if err := ledgerValues(u.Minutes, u.CostMicro); err != nil { + return err + } + if err := putUsage(ctx, tx, u); err != nil { + return err + } + } + return ledgerTx(s, ctx, tx).pruneLedger() + }) +} + // AllowanceInputs are the figures an allowance is made of for one provider and UTC month. type AllowanceInputs struct { // Imported is the provider-reported consumption of the month (saturating); HasImport says there is any.