agent (unreleased): a refused signed job and a watchdog give-up reach the hub (R-1383, D78)
gates / gates (push) Successful in 42s

The job runner clears a job with its outcome, a closed reason code and the op's name. The guest-power watchdog's
give-up rides the host report until the guest runs.

Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS
This commit is contained in:
admin committed 2026-11-14 17:29:53 +01:00
1 parent ede5d0e6c7
commit a1135018c7
11 files changed
+327 -15

No files matched your search

+10
View File
@@ -2,6 +2,16 @@
Commit only. Each part can be released alone; none needs release 2b. No `configs/` change in this section.
- **A refused signed job and a watchdog give-up reach the hub (R-1383, D78, felhom.eu `09` §3 decision 299; `05`
§16.6).** (1) `signedjobs.Runner` clears a job with `hub.Client.CompleteJobWithOutcome`: `outcome` (done · rejected
· failed · malformed), a `reason` from the closed `signedjobs.ReasonCodes` (the two clock reasons `expired` and
`not_yet_valid` first — a box with a wrong clock refuses every job) and the op's name, in the `DELETE`'s query. No
signature, nonce or job parameter is sent. (2) The guest-power watchdog's give-up — one ERROR line in the journal
until now — rides the host report as `guest_power_gave_up` (`localapi.Server.GuestPowerGaveUpStanza`, wired in
`main.go`); the list empties when the guest runs. Tests `TestRunner_TellsTheHubHowEachJobEnded` (the real gate
and verifier), `TestReasonCode_ClosedList`, `TestCompleteJob_OutcomeRidesTheQuery`,
`TestGuestPower_GiveUpIsReportedUntilTheGuestRuns`, `TestMainWiresPVETrust`; 12 mutations seen red (felhom.eu
`documentation/audits/night-build-2026-11-14/B/redproof-agent-{job-outcome,guest-power}.txt`).
- **The report carries the box's clock at the moment of sending (R-1384, D79, felhom.eu `09` §3 decision 300).**
`HostReport.SentAt` (`sent_at`, RFC3339 with nanoseconds) is stamped by `Client.Report` at every send, never by the
collector. `reported_at` could not serve: it is stamped when the collect starts. The hub (its own unreleased part)
+4
View File
@@ -1369,6 +1369,10 @@ func runDaemon(cfg config.Config, logger *slog.Logger, logRing *applog.Ring) int
},
}
localSrv := buildLocalAPIServer(cfg, px, backupStore, heavyOps, observer, driveKnown, hostOps, gate, collector, client, intentRec, guestBindStore, formatJobStore, logRing, escrowCeremonyCfg, logger, &localTokens)
if localSrv != nil {
// R-1383 (D78): the guest-power watchdog's give-up rides the host report (it was one journal line).
collector.SetGuestPowerGaveUpSource(localSrv.GuestPowerGaveUpStanza)
}
if localSrv != nil {
// 2026-10-29 (operator brief, Part 1): the operator's actions for THIS agent ride the host report's reply — the
// closed list (localapi/opactions.go). The server joins the observer chain; its results ride the next report,
+6 -1
View File
@@ -31,6 +31,10 @@ func TestMainWiresPVETrust(t *testing.T) {
seen["ensure"]++
case sel.Sel.Name == "SetProxmoxTrustSource":
seen["stanza"]++
case sel.Sel.Name == "SetGuestPowerGaveUpSource" && len(node.Args) == 1:
if a, ok := node.Args[0].(*ast.SelectorExpr); ok && a.Sel.Name == "GuestPowerGaveUpStanza" {
seen["guest-power give-up → report"]++ // R-1383
}
case sel.Sel.Name == "SetCollectErrorHook" && len(node.Args) == 1:
if a, ok := node.Args[0].(*ast.SelectorExpr); ok && a.Sel.Name == "OnCallError" {
seen["hook"]++
@@ -53,7 +57,8 @@ func TestMainWiresPVETrust(t *testing.T) {
}
return true
})
want := map[string]int{"ensure": 2, "stanza": 1, "hook": 1, "timer": 1, "apply=SetTLS": 1, "profile-refresh": 2}
want := map[string]int{"ensure": 2, "stanza": 1, "hook": 1, "timer": 1, "apply=SetTLS": 1, "profile-refresh": 2,
"guest-power give-up → report": 1}
for k, n := range want {
if seen[k] < n {
t.Errorf("main.go wires %q %d time(s), want at least %d (the two-call ones: at start and after a bundle install)", k, seen[k], n)
+29 -1
View File
@@ -10,6 +10,7 @@ import (
"io"
"log/slog"
"net/http"
"net/url"
"os"
"strings"
"time"
@@ -286,10 +287,37 @@ func (c *Client) Jobs(ctx context.Context) ([]JobWire, error) {
// /hosts/{id}/jobs/{job_id}, self-scoped. Called after a job is executed OR permanently rejected
// (the nonce is already durably burned on a passing verify, so re-processing is replay-safe).
func (c *Client) CompleteJob(ctx context.Context, jobID string) error {
return c.CompleteJobWithOutcome(ctx, jobID, JobOutcome{})
}
// JobOutcome says HOW a signed job ended when the agent clears it (R-1383, D78): Outcome is done | rejected | failed
// | malformed; Reason is one code of signedjobs.ReasonCodes (empty for done); Op is the job's op class. It rides the
// DELETE's query, so a hub that does not know it ignores it. Until this, „ran” and „refused” looked the same to the
// hub, and a refused agent update left no trace outside the box's journal. No secret: no signature, no nonce, no
// parameter of the job is in it.
type JobOutcome struct {
Outcome string
Reason string
Op string
}
// CompleteJobWithOutcome clears the job and tells the hub how it ended. The zero JobOutcome sends no query.
func (c *Client) CompleteJobWithOutcome(ctx context.Context, jobID string, o JobOutcome) error {
if c.hostID == "" || jobID == "" {
return fmt.Errorf("hub: CompleteJob requires host_id + job_id")
}
req, err := http.NewRequestWithContext(ctx, http.MethodDelete, c.baseURL+"/api/v1/hosts/"+c.hostID+"/jobs/"+jobID, nil)
u := c.baseURL + "/api/v1/hosts/" + c.hostID + "/jobs/" + jobID
if o.Outcome != "" {
q := url.Values{"outcome": {o.Outcome}}
if o.Reason != "" {
q.Set("reason", o.Reason)
}
if o.Op != "" {
q.Set("op", o.Op)
}
u += "?" + q.Encode()
}
req, err := http.NewRequestWithContext(ctx, http.MethodDelete, u, nil)
if err != nil {
return err
}
+23
View File
@@ -51,3 +51,26 @@ func TestReport_StampsSentAtAtSending(t *testing.T) {
t.Errorf("the second send carries %s, not after the first %s — the stamp is not taken per send", sent[1], sent[0])
}
}
// R-1383 (D78): the clearing call says how the job ended, in the DELETE's query; a plain CompleteJob sends none (an
// older hub, and the „nothing to say” case, see the same request as before).
func TestCompleteJob_OutcomeRidesTheQuery(t *testing.T) {
var urls []string
c := testClient(func(r *http.Request) (*http.Response, error) {
if r.Method != http.MethodDelete {
t.Errorf("method = %s", r.Method)
}
urls = append(urls, r.URL.Path+"?"+r.URL.RawQuery)
return httpResp(200, `{"status":"ok"}`), nil
})
if err := c.CompleteJob(context.Background(), "j1"); err != nil {
t.Fatal(err)
}
if err := c.CompleteJobWithOutcome(context.Background(), "j2", JobOutcome{Outcome: "rejected", Reason: "expired", Op: "agent_update"}); err != nil {
t.Fatal(err)
}
want := []string{"/api/v1/hosts/demo-host-01/jobs/j1?", "/api/v1/hosts/demo-host-01/jobs/j2?op=agent_update&outcome=rejected&reason=expired"}
if len(urls) != 2 || urls[0] != want[0] || urls[1] != want[1] {
t.Fatalf("requests = %q, want %q", urls, want)
}
}
+22 -7
View File
@@ -125,6 +125,7 @@ type Collector struct {
capProbe func(ctx context.Context) []capability.Status // v0.44.0: privileged-capability self-check (nil → empty)
leafFP string // v0.48.0: served local-API leaf fp (static per process; "" when local API disabled)
pveTrust func() any // R-1380: the proxmox_trust stanza (nil = omitted)
guestPowerGaveUp func() any // R-1383: the guest_power_gave_up stanza (nil = omitted)
addrEnum AddressEnumerator // v0.119.0: host interface enumeration; nil => the REAL one (see collectAddresses)
wg WireguardReporter // S3: offsite-tunnel status (nil → stanza omitted)
pbsdr PBSDRReporter // slice 2: PBS DR tier bridge state (nil → stanza omitted)
@@ -216,6 +217,12 @@ func (c *Collector) SetProxmoxTrustSource(f func() any) *Collector {
return c
}
// SetGuestPowerGaveUpSource wires the `guest_power_gave_up` stanza (R-1383). Optional; nil, or a nil result, omits it.
func (c *Collector) SetGuestPowerGaveUpSource(f func() any) *Collector {
c.guestPowerGaveUp = f
return c
}
// SetLeafFingerprint records the served local-API leaf fp (v0.48.0) to ride every host report (the hub
// watches it for a re-key). Static per process — set once at startup. "" when the local API is
// disabled. Returns the collector for chaining.
@@ -398,6 +405,13 @@ func (c *Collector) SetOOBReporter(o OOBReporter) *Collector {
// error (no useful report — the cycle skips the POST); a failed per-guest
// GuestConfig degrades that guest to status="unknown" without spec but still sends;
// a cloudflared probe failure yields status="unknown" and is never fatal.
func (c *Collector) guestPowerGaveUpStanza() any {
if c.guestPowerGaveUp == nil {
return nil
}
return c.guestPowerGaveUp()
}
func (c *Collector) proxmoxTrust() any {
if c.pveTrust == nil {
return nil
@@ -430,13 +444,14 @@ func (c *Collector) Collect(ctx context.Context) (*HostReport, error) {
RestoreTests: c.collectRestoreTests(ctx),
PBSSnapshots: c.collectPBSSnapshots(ctx),
AuditTail: []AuditEntry{},
Cloudflared: c.cloudflared(ctx),
Capabilities: c.capabilities(ctx),
LeafFingerprint: c.leafFP,
ProxmoxTrust: c.proxmoxTrust(),
Addresses: c.collectAddresses(),
System: c.systemInfo(ctx, ns),
AuditTail: []AuditEntry{},
Cloudflared: c.cloudflared(ctx),
Capabilities: c.capabilities(ctx),
LeafFingerprint: c.leafFP,
ProxmoxTrust: c.proxmoxTrust(),
GuestPowerGaveUp: c.guestPowerGaveUpStanza(),
Addresses: c.collectAddresses(),
System: c.systemInfo(ctx, ns),
}
// DR recipe host-half — derived from the just-collected guest/storage/PBS facts (no new reads).
// Secret-free by construction (identifiers/intents/sizes/coordinates only).
+4
View File
@@ -182,6 +182,10 @@ type HostReport struct {
// byte-stable. The hub parser is pinned by hub/internal/monitor/controller_supervisor_test.go
// against the JSON TestControllerSupervisorStanza_WireShape pins here.
ControllerSupervisor *ControllerSupervisorStatus `json:"controller_supervisor,omitempty"`
// GuestPowerGaveUp (R-1383, D78): the guests the guest-power watchdog stopped trying to start (localapi
// GuestPowerGiveUp, field-exact: vmid, since, attempts). The hub raises one operator alarm per guest and clears it
// when the guest runs. `any` + omitempty: absent when there is none or the source is not wired.
GuestPowerGaveUp any `json:"guest_power_gave_up,omitempty"`
// OperatorActionResults (2026-10-29, the operator brief's Part 1): the answers to the operator actions the hub's
// replies listed — the controller's shape (`04` §6.1). Re-sent until the reply stops listing the id. omitempty: absent
// when there is nothing to answer, so the golden stays byte-stable.
+36
View File
@@ -3,6 +3,7 @@ package localapi
import (
"context"
"log/slog"
"sort"
"sync"
"time"
@@ -82,6 +83,40 @@ type guestPowerState struct {
attempts int
nextAt time.Time
raised bool // the give-up fault has already been raised for this run
raisedAt time.Time
}
// GuestPowerGiveUp is one guest the watchdog stopped trying to start (R-1383, D78). It rides the host report
// (`guest_power_gave_up`) so the operator hears it: until this, the give-up was one ERROR line in the host's journal
// while the household's apps were off. Memory only — an agent restart forgets it (and tries three times again).
type GuestPowerGiveUp struct {
VMID int `json:"vmid"`
Since string `json:"since"` // RFC3339 UTC, when the watchdog gave up
Attempts int `json:"attempts"`
}
// GuestPowerGaveUpStanza is GuestPowerGaveUp shaped for the host report: an UNTYPED nil when there is none, so the
// report omits the field (a typed nil slice in an `any` would be sent as `null`).
func (s *Server) GuestPowerGaveUpStanza() any {
if g := s.GuestPowerGaveUp(); len(g) > 0 {
return g
}
return nil
}
// GuestPowerGaveUp returns the guests the watchdog has given up on, by vmid. nil when there is none — the report
// stanza is then omitted. Pinned by TestGuestPower_GiveUpIsReportedUntilTheGuestRuns.
func (s *Server) GuestPowerGaveUp() []GuestPowerGiveUp {
guestPowerMu.Lock()
defer guestPowerMu.Unlock()
var out []GuestPowerGiveUp
for vmid, st := range s.guestPower {
if st.raised {
out = append(out, GuestPowerGiveUp{VMID: vmid, Since: st.raisedAt.UTC().Format(time.RFC3339), Attempts: st.attempts})
}
}
sort.Slice(out, func(i, j int) bool { return out[i].VMID < out[j].VMID })
return out
}
// WatchGuestPower runs the guest-power sweep every guestPowerInterval until ctx is done. No-op when
@@ -260,6 +295,7 @@ func (s *Server) markGuestPowerRaised(vmid int) {
defer guestPowerMu.Unlock()
st := s.guestPower[vmid]
st.raised = true
st.raisedAt = s.now()
s.guestPower[vmid] = st
}
+44
View File
@@ -2,6 +2,7 @@ package localapi
import (
"context"
"encoding/json"
"errors"
"log/slog"
"testing"
@@ -237,3 +238,46 @@ func TestGuestPower_RunningGuestClearsAttemptHistory(t *testing.T) {
t.Error("attempt history survived the guest coming back healthy")
}
}
// R-1383 (D78): the give-up was ONE journal line while the household's apps were off. It now rides the host report:
// nothing before the give-up (and an untyped nil, so the report omits the field), the guest with the time and the
// attempt count after it, and nothing again once the guest runs.
// COMPANION RED-PROOFS (observed, felhom.eu audits/night-build-2026-11-14/B/redproof-agent-guest-power.txt).
func TestGuestPower_GiveUpIsReportedUntilTheGuestRuns(t *testing.T) {
ctl := &fakeGuestPowerCtl{
guests: []proxmox.Guest{{VMID: 9201, Status: "stopped"}},
locks: map[int]string{9201: ""},
onboot: map[int]bool{9201: true},
startErr: errors.New("cannot start: storage offline"),
}
now := time.Date(2026, 11, 14, 2, 0, 0, 0, time.UTC)
s := gpServer(t, ctl, func() time.Time { return now })
if st := s.GuestPowerGaveUpStanza(); st != nil {
t.Fatalf("before any attempt the stanza is %#v, want an untyped nil (the report omits the field)", st)
}
for i := 0; i < guestPowerMaxAttempts; i++ { // the three failed starts
s.GuestPowerTick(context.Background())
if g := s.GuestPowerGaveUp(); len(g) != 0 {
t.Fatalf("after %d failed start(s) the watchdog still tries, but it reports a give-up: %+v", i+1, g)
}
now = now.Add(10 * time.Minute)
}
gaveUpAt := now
s.GuestPowerTick(context.Background()) // the tick that gives up
now = now.Add(time.Hour)
s.GuestPowerTick(context.Background())
g := s.GuestPowerGaveUp()
if len(g) != 1 || g[0].VMID != 9201 || g[0].Attempts != guestPowerMaxAttempts || g[0].Since != gaveUpAt.Format(time.RFC3339) {
t.Fatalf("give-up list = %+v, want guest 9201, %d attempts, since %s (the give-up's time, not a later tick's)", g, guestPowerMaxAttempts, gaveUpAt.Format(time.RFC3339))
}
b, _ := json.Marshal(s.GuestPowerGaveUpStanza())
if string(b) != `[{"vmid":9201,"since":"2026-11-14T02:30:00Z","attempts":3}]` {
t.Fatalf("the stanza on the wire = %s", b)
}
// the operator starts the guest: the next sweep sees it running and the list is empty again
ctl.guests = []proxmox.Guest{{VMID: 9201, Status: "running"}}
s.GuestPowerTick(context.Background())
if st := s.GuestPowerGaveUpStanza(); st != nil {
t.Fatalf("the guest runs again and the stanza is still %#v", st)
}
}
+101
View File
@@ -0,0 +1,101 @@
package signedjobs
import (
"context"
"errors"
"sync"
"testing"
"time"
"gitea.dooplex.hu/admin/felhom-agent/internal/hub"
"gitea.dooplex.hu/admin/felhom-agent/internal/reconcile"
)
// outcomeQueue is a jobsQueue that also has the optional half (OutcomeSource), as *hub.Client does.
type outcomeQueue struct {
jobsQueue
omu sync.Mutex
outcomes map[string]hub.JobOutcome
}
func (q *outcomeQueue) CompleteJobWithOutcome(ctx context.Context, jobID string, o hub.JobOutcome) error {
q.omu.Lock()
if q.outcomes == nil {
q.outcomes = map[string]hub.JobOutcome{}
}
q.outcomes[jobID] = o
q.omu.Unlock()
return q.jobsQueue.CompleteJob(ctx, jobID)
}
// R-1383 (D78): a signed job that is refused or fails was cleared from the hub's queue exactly like one that ran —
// „an agent update never arrives” left one journal line on the box. The runner now says how each job ended, through
// the REAL gate and verifier: the stored effect is the outcome the hub receives with the clearing call.
// The two clock reasons are the ones the ruling is about: a box with a wrong clock refuses every job.
// COMPANION RED-PROOFS (observed, felhom.eu audits/night-build-2026-11-14/B/redproof-agent-job-outcome.txt).
func TestRunner_TellsTheHubHowEachJobEnded(t *testing.T) {
s := newTestSigner(t)
_, _, exec := newRealGateRunner(t, s)
base, _, _ := newRealGateRunner(t, s)
src := &outcomeQueue{}
r := NewRunner(src, base.gate, exec, testHost, quiet())
now := time.Now().UTC()
src.add(mintJob(t, s, "ok", testHost, "", "ops-1", wipeParamsJSON, now, now.Add(time.Hour)))
src.add(mintJob(t, s, "expired", testHost, "", "ops-1", wipeParamsJSON, now.Add(-3*time.Hour), now.Add(-2*time.Hour)))
src.add(mintJob(t, s, "early", testHost, "", "ops-1", wipeParamsJSON, now.Add(2*time.Hour), now.Add(3*time.Hour)))
src.add(mintJob(t, s, "other-box", "some-other-host", "", "ops-1", wipeParamsJSON, now, now.Add(time.Hour)))
src.add(corruptSig(t, mintJob(t, s, "forged", testHost, "", "ops-1", wipeParamsJSON, now, now.Add(time.Hour))))
src.add(hub.JobWire{JobID: "garbage", BlobB64: "!!!not-base64!!!"})
if _, err := r.RunOnce(context.Background()); err != nil {
t.Fatal(err)
}
// and a job that is accepted and whose run fails
exec.err = errors.New("the wrapper refused")
src.add(mintJob(t, s, "run-fails", testHost, "", "ops-1", wipeParamsJSON, now, now.Add(time.Hour)))
if _, err := r.RunOnce(context.Background()); err != nil {
t.Fatal(err)
}
want := map[string]hub.JobOutcome{
"ok": {Outcome: "done", Op: "storage_wipe"},
"expired": {Outcome: "rejected", Reason: "expired", Op: "storage_wipe"},
"early": {Outcome: "rejected", Reason: "not_yet_valid", Op: "storage_wipe"},
"other-box": {Outcome: "rejected", Reason: "target", Op: "storage_wipe"},
"forged": {Outcome: "rejected", Reason: "malformed", Op: "storage_wipe"}, // corruptSig breaks the armour itself
"garbage": {Outcome: "malformed", Reason: "malformed"},
"run-fails": {Outcome: "failed", Reason: "execution_failed", Op: "storage_wipe"},
}
for id, w := range want {
if got, ok := src.outcomes[id]; !ok || got != w {
t.Errorf("job %q: the hub was told %+v (told=%v), want %+v", id, got, ok, w)
}
}
if len(src.outcomes) != len(want) {
t.Errorf("outcomes for %d jobs, want %d: %+v", len(src.outcomes), len(want), src.outcomes)
}
if exec.count() != 2 {
t.Errorf("executor ran %d times, want 2 (ok and run-fails only)", exec.count())
}
}
// Every code reasonCode can return is on the closed list the hub holds a sentence for.
func TestReasonCode_ClosedList(t *testing.T) {
known := map[string]bool{}
for _, c := range ReasonCodes {
known[c] = true
}
for _, d := range []reconcile.Decision{
{}, {Reason: reconcile.ReasonRejected, Err: errors.New("something new")},
{Reason: reconcile.ReasonRoleDenied}, {Reason: reconcile.ReasonBindingMismatch}, {Reason: reconcile.ReasonPendingSignature},
} {
if c := reasonCode(d); !known[c] {
t.Errorf("reasonCode(%+v) = %q, not on ReasonCodes", d, c)
}
}
if reasonCode(reconcile.Decision{Reason: reconcile.ReasonRejected, Err: errors.New("x")}) != "other" {
t.Error("an unknown refusal must be told as „other”, never as the error's own text")
}
if !known["execution_failed"] || !known["malformed"] {
t.Error("the two codes the runner sets itself are not on the list")
}
}
+48 -6
View File
@@ -35,6 +35,42 @@ type JobSource interface {
CompleteJob(ctx context.Context, jobID string) error
}
// OutcomeSource is the optional half of a JobSource: it clears a job AND tells the hub how it ended (R-1383, D78).
// *hub.Client has it. A source without it is cleared as before.
type OutcomeSource interface {
CompleteJobWithOutcome(ctx context.Context, jobID string, o hub.JobOutcome) error
}
// ReasonCodes is the CLOSED list of reasons the agent sends with a job that did not run to a good end. The hub holds
// one operator sentence per code (felhom.eu hub/internal/api, signedJobReasons); a code the hub does not know is
// told as „other”. Pinned by TestReasonCode_ClosedList.
var ReasonCodes = []string{"expired", "not_yet_valid", "window_too_long", "replay", "bad_signature", "unknown_signer",
"target", "namespace", "malformed", "role_denied", "binding_mismatch", "execution_failed", "other"}
// reasonCode names why the gate refused: the authz error first (the two clock reasons matter most — a box with a
// wrong clock refuses EVERY job), then the gate's own reason.
func reasonCode(dec reconcile.Decision) string {
for _, m := range []struct {
err error
code string
}{
{authz.ErrExpired, "expired"}, {authz.ErrNotYetValid, "not_yet_valid"}, {authz.ErrWindowTooLong, "window_too_long"},
{authz.ErrReplay, "replay"}, {authz.ErrBadSignature, "bad_signature"}, {authz.ErrUnknownSigner, "unknown_signer"},
{authz.ErrTarget, "target"}, {authz.ErrNamespace, "namespace"}, {authz.ErrMalformed, "malformed"},
} {
if errorsIs(dec.Err, m.err) {
return m.code
}
}
switch dec.Reason {
case reconcile.ReasonRoleDenied:
return "role_denied"
case reconcile.ReasonBindingMismatch:
return "binding_mismatch"
}
return "other"
}
// Authorizer is the reversibility gate. Satisfied by *reconcile.Gate.
type Authorizer interface {
Authorize(intent reconcile.Intent, signed *reconcile.SignedOp) reconcile.Decision
@@ -141,7 +177,7 @@ func (r *Runner) processJob(ctx context.Context, j hub.JobWire) bool {
// A malformed/forged envelope is permanently bad — clear it loudly (a queued garbage job
// must not wedge the queue). No nonce was consumed (we never reached Verify).
r.logger.Error("signedjobs: malformed job envelope — clearing", "job", j.JobID)
r.complete(ctx, j.JobID)
r.complete(ctx, j.JobID, hub.JobOutcome{Outcome: "malformed", Reason: "malformed"})
return true
}
@@ -166,7 +202,7 @@ func (r *Runner) processJob(ctx context.Context, j hub.JobWire) bool {
// it as a security event (a compromised hub queuing a forged op lands here).
r.logger.Warn("signedjobs: REJECTED signed op — executor not called",
"job", j.JobID, "op", ob.Op, "reason", dec.Reason, "err", errStr(dec.Err))
r.complete(ctx, j.JobID)
r.complete(ctx, j.JobID, hub.JobOutcome{Outcome: "rejected", Reason: reasonCode(dec), Op: ob.Op})
return true
}
@@ -177,7 +213,7 @@ func (r *Runner) processJob(ctx context.Context, j hub.JobWire) bool {
switch {
case err == nil:
r.logger.Warn("signedjobs: signed op COMPLETED", "job", j.JobID, "op", ob.Op)
r.complete(ctx, j.JobID)
r.complete(ctx, j.JobID, hub.JobOutcome{Outcome: "done", Op: ob.Op})
return true
case errorsIs(err, ErrNoExecutor):
// No executor for this op class in this build — leave it queued for the owning slice.
@@ -189,7 +225,7 @@ func (r *Runner) processJob(ctx context.Context, j hub.JobWire) bool {
// nonce to retry) and log the failure loudly.
r.logger.Error("signedjobs: signed op execution FAILED (nonce spent — clearing)",
"job", j.JobID, "op", ob.Op, "err", err)
r.complete(ctx, j.JobID)
r.complete(ctx, j.JobID, hub.JobOutcome{Outcome: "failed", Reason: "execution_failed", Op: ob.Op})
return true
}
}
@@ -215,8 +251,14 @@ func (r *Runner) decode(j hub.JobWire) (*reconcile.SignedOp, authz.OpBlob, bool)
return &reconcile.SignedOp{Blob: opBlob, Sig: []byte(env.SigArmored)}, ob, true
}
func (r *Runner) complete(ctx context.Context, jobID string) {
if err := r.source.CompleteJob(ctx, jobID); err != nil {
func (r *Runner) complete(ctx context.Context, jobID string, o hub.JobOutcome) {
var err error
if os, ok := r.source.(OutcomeSource); ok {
err = os.CompleteJobWithOutcome(ctx, jobID, o) // R-1383: the hub hears HOW the job ended
} else {
err = r.source.CompleteJob(ctx, jobID)
}
if err != nil {
r.logger.Warn("signedjobs: clearing job failed (will retry next pass)", "job", jobID, "err", err)
}
}