From d20c93cbf97591aaa54f5f2f5b10118e2a6bc0bd Mon Sep 17 00:00:00 2001 From: Igor Date: Mon, 19 Oct 2026 08:24:30 -0500 Subject: [PATCH 1/2] abandon-stuck: hand back this worker's in-flight backlog with JobRegistry.abandonJob The worker now abandons a job it fails, but a backlog from before that change stays Acknowledged until the keeper's no-fault settle 48 h after each deadline. At maxConcurrentJobs of those the chain refuses the worker's session claims, and WorkerRegistry.removeSupportedModel refuses while any job for the model is still counted active, so the backlog blocks both recovery paths (g1 worker-4/5, 2026-10-19). `lightchain-worker abandon-stuck` finds this worker's jobs through JobSubmitted events in the last ABANDON_LOOKBACK_BLOCKS (default 150000), reads each state, and calls abandonJob for every one still Submitted or Acknowledged: consumer refunded, in-flight slot released. ABANDON_DRY_RUN=1 only lists. A failed abandon is reported and skipped, never fatal. --- cmd/cli/main.go | 48 +++++++++++ internal/cli/abandon_stuck.go | 129 +++++++++++++++++++++++++++++ internal/cli/abandon_stuck_test.go | 110 ++++++++++++++++++++++++ 3 files changed, 287 insertions(+) create mode 100644 internal/cli/abandon_stuck.go create mode 100644 internal/cli/abandon_stuck_test.go diff --git a/cmd/cli/main.go b/cmd/cli/main.go index 25fc1ec..68c39f9 100644 --- a/cmd/cli/main.go +++ b/cmd/cli/main.go @@ -34,6 +34,7 @@ import ( "log/slog" "os" "slices" + "strconv" "strings" "time" @@ -80,6 +81,8 @@ func main() { runAddModels() case "remove-models": runRemoveModels() + case "abandon-stuck": + runAbandonStuck() case "reinstate": runReinstate() case "top-up-stake": @@ -125,6 +128,14 @@ Commands: keygen Generate/load ECDH encryption key and print public key hex register Register worker on-chain (stake + encryption key + models) add-models Add models to an already-registered worker + remove-models + Remove the models named in SUPPORTED_MODELS from the worker's + on-chain list (sortition routes by the registry, not the env) + abandon-stuck + Hand back every job of this worker still Submitted/Acknowledged + on chain (JobRegistry.abandonJob): refunds the consumer, frees the + in-flight slot. ABANDON_LOOKBACK_BLOCKS (default 150000) bounds + the scan, ABANDON_DRY_RUN=1 only lists. drain Mark worker ineligible for new sessions (selection-only) undrain Reverse drain — restore worker eligibility deregister Deregister worker and withdraw stake @@ -347,6 +358,43 @@ func runRemoveModels() { } } +// runAbandonStuck sweeps this worker's in-flight backlog (see cli.AbandonStuck). +// Each abandon is one transaction, so the budget is txTimeout per job plus the scan. +func runAbandonStuck() { + cfg, logger := loadAndValidateCfg() + signingKey, workerAddr := loadSigningKey(cfg, logger) + + lookback := uint64(150_000) + if v := os.Getenv("ABANDON_LOOKBACK_BLOCKS"); v != "" { + n, err := strconv.ParseUint(v, 10, 64) + if err != nil { + logger.Error("ABANDON_LOOKBACK_BLOCKS must be a whole number of blocks", "value", v) + os.Exit(2) + } + lookback = n + } + dryRun := os.Getenv("ABANDON_DRY_RUN") == "1" + + chainClient := dialChain(cfg, signingKey, logger) + defer chainClient.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) + defer cancel() + + n, err := cli.AbandonStuck(ctx, cli.AbandonStuckOpts{ + Client: chainClient, + Worker: workerAddr, + LookbackBlocks: lookback, + DryRun: dryRun, + Out: os.Stdout, + Logger: logger, + }) + if err != nil { + logger.Error("abandon-stuck finished with errors", "abandoned", n, "error", err) + os.Exit(1) + } +} + func runDeregister() { cfg, logger := loadAndValidateCfg() signingKey, workerAddr := loadSigningKey(cfg, logger) diff --git a/internal/cli/abandon_stuck.go b/internal/cli/abandon_stuck.go new file mode 100644 index 0000000..c15fb3d --- /dev/null +++ b/internal/cli/abandon_stuck.go @@ -0,0 +1,129 @@ +package cli + +import ( + "context" + "fmt" + "io" + "log/slog" + + "github.com/ethereum/go-ethereum/common" + + "github.com/lightchain/worker/internal/chain" +) + +// StuckChain is the narrow chain surface the abandon-stuck command needs. +type StuckChain interface { + Head(ctx context.Context) (chain.HeadInfo, error) + FilterJobSubmitted(ctx context.Context, fromBlock, toBlock uint64) ([]chain.JobSubmittedEvent, error) + GetJobState(ctx context.Context, jobID uint64) (chain.JobStateInfo, error) + AbandonJob(ctx context.Context, jobID uint64) error +} + +// AbandonStuckOpts configures one backlog sweep. +type AbandonStuckOpts struct { + Client StuckChain + Worker common.Address + LookbackBlocks uint64 // how far behind the head to look for this worker's jobs + ChunkSize uint64 // eth_getLogs range per request (RPC limits), default 5000 + DryRun bool // list, do not send + Out io.Writer + Logger *slog.Logger +} + +// AbandonStuck hands back every job of this worker that is still Submitted or +// Acknowledged on chain (JobRegistry.abandonJob): the consumer is refunded and +// the job leaves the worker's in-flight count at once. +// +// Why a separate command: the worker abandons a job it fails from now on, but a +// backlog from before that change stays Acknowledged until the keeper's +// no-fault settle 48 h after each deadline. At maxConcurrentJobs of those the +// chain refuses the worker's session claims (2026-10-19, g1 worker-4/5), and +// WorkerRegistry.removeSupportedModel refuses while any job for the model is +// still counted active, so the backlog blocks both recovery paths. +// +// Jobs are found through JobSubmitted events in the look-back window; anything +// older has been swept by the keeper already. Each candidate's state is read +// before the transaction, and a failed abandon is reported and skipped, never +// fatal, so one odd job cannot stop the sweep. +func AbandonStuck(ctx context.Context, o AbandonStuckOpts) (abandoned int, err error) { + if o.Client == nil { + return 0, fmt.Errorf("abandon-stuck: no chain client") + } + chunk := o.ChunkSize + if chunk == 0 { + chunk = 5000 + } + log := o.Logger + if log == nil { + log = slog.Default() + } + + head, err := o.Client.Head(ctx) + if err != nil { + return 0, fmt.Errorf("abandon-stuck: head: %w", err) + } + from := uint64(1) + if head.Number > o.LookbackBlocks { + from = head.Number - o.LookbackBlocks + } + + // Phase 1: this worker's jobs in the window. + var mine []uint64 + for lo := from; lo <= head.Number; lo += chunk { + hi := lo + chunk - 1 + if hi > head.Number { + hi = head.Number + } + evs, err := o.Client.FilterJobSubmitted(ctx, lo, hi) + if err != nil { + return 0, fmt.Errorf("abandon-stuck: JobSubmitted %d-%d: %w", lo, hi, err) + } + for _, ev := range evs { + if ev.Worker == o.Worker { + mine = append(mine, ev.JobID) + } + } + } + fmt.Fprintf(o.Out, "worker %s: %d jobs submitted in blocks %d-%d\n", o.Worker.Hex(), len(mine), from, head.Number) + + // Phase 2: the ones still in flight. + var stuck []uint64 + for _, id := range mine { + st, err := o.Client.GetJobState(ctx, id) + if err != nil { + log.Warn("getJob failed; skipping", "jobID", id, "error", err) + continue + } + if st.State == chain.JobStateSubmitted || st.State == chain.JobStateAcknowledged { + stuck = append(stuck, id) + } + } + fmt.Fprintf(o.Out, "in flight (Submitted/Acknowledged): %d\n", len(stuck)) + if len(stuck) == 0 { + return 0, nil + } + if o.DryRun { + fmt.Fprintf(o.Out, "dry run: would abandon %v\n", stuck) + return 0, nil + } + + // Phase 3: hand them back, one tx each. + var failed []uint64 + for _, id := range stuck { + if err := ctx.Err(); err != nil { + return abandoned, fmt.Errorf("abandon-stuck: stopped after %d of %d: %w", abandoned, len(stuck), err) + } + if err := o.Client.AbandonJob(ctx, id); err != nil { + log.Warn("abandonJob failed; skipping", "jobID", id, "error", err) + failed = append(failed, id) + continue + } + abandoned++ + log.Info("job abandoned: fee refunded, in-flight slot released", "jobID", id) + } + fmt.Fprintf(o.Out, "abandoned %d of %d\n", abandoned, len(stuck)) + if len(failed) > 0 { + return abandoned, fmt.Errorf("%d job(s) could not be abandoned (already settled, or the tx reverted): %v", len(failed), failed) + } + return abandoned, nil +} diff --git a/internal/cli/abandon_stuck_test.go b/internal/cli/abandon_stuck_test.go new file mode 100644 index 0000000..57cd06f --- /dev/null +++ b/internal/cli/abandon_stuck_test.go @@ -0,0 +1,110 @@ +package cli + +import ( + "bytes" + "context" + "errors" + "testing" + + "github.com/ethereum/go-ethereum/common" + "github.com/stretchr/testify/require" + + "github.com/lightchain/worker/internal/chain" +) + +type stuckChain struct { + head uint64 + events []chain.JobSubmittedEvent + states map[uint64]chain.JobStateInfo + abandoned []uint64 + failOn uint64 + ranges [][2]uint64 +} + +func (s *stuckChain) Head(context.Context) (chain.HeadInfo, error) { + return chain.HeadInfo{Number: s.head}, nil +} + +func (s *stuckChain) FilterJobSubmitted(_ context.Context, from, to uint64) ([]chain.JobSubmittedEvent, error) { + s.ranges = append(s.ranges, [2]uint64{from, to}) + var out []chain.JobSubmittedEvent + for _, ev := range s.events { + if ev.BlockNumber >= from && ev.BlockNumber <= to { + out = append(out, ev) + } + } + return out, nil +} + +func (s *stuckChain) GetJobState(_ context.Context, id uint64) (chain.JobStateInfo, error) { + st, ok := s.states[id] + if !ok { + return chain.JobStateInfo{}, errors.New("no such job") + } + return st, nil +} + +func (s *stuckChain) AbandonJob(_ context.Context, id uint64) error { + if id == s.failOn { + return errors.New("execution reverted") + } + s.abandoned = append(s.abandoned, id) + return nil +} + +var ( + me = common.HexToAddress("0x00000000000000000000000000000000000000a1") + other = common.HexToAddress("0x00000000000000000000000000000000000000b2") +) + +func newStuckChain() *stuckChain { + return &stuckChain{ + head: 10_000, + events: []chain.JobSubmittedEvent{ + {JobID: 1, Worker: me, BlockNumber: 100}, // outside the look-back + {JobID: 2, Worker: me, BlockNumber: 6_000}, // acknowledged, mine + {JobID: 3, Worker: other, BlockNumber: 6_100}, + {JobID: 4, Worker: me, BlockNumber: 7_000}, // completed + {JobID: 5, Worker: me, BlockNumber: 9_900}, // submitted, mine + }, + states: map[uint64]chain.JobStateInfo{ + 1: {State: chain.JobStateAcknowledged, Worker: me}, + 2: {State: chain.JobStateAcknowledged, Worker: me}, + 3: {State: chain.JobStateAcknowledged, Worker: other}, + 4: {State: chain.JobStateCompleted, Worker: me}, + 5: {State: chain.JobStateSubmitted, Worker: me}, + }, + } +} + +func TestAbandonStuck_AbandonsOnlyMyInFlightJobsInWindow(t *testing.T) { + c := newStuckChain() + var out bytes.Buffer + n, err := AbandonStuck(context.Background(), AbandonStuckOpts{Client: c, Worker: me, LookbackBlocks: 5_000, ChunkSize: 2_000, Out: &out}) + require.NoError(t, err) + require.Equal(t, 2, n) + require.Equal(t, []uint64{2, 5}, c.abandoned, "job 1 is outside the window, 3 belongs to another worker, 4 is completed") + require.Equal(t, [2]uint64{5_000, 6_999}, c.ranges[0], "scan starts head - lookback and respects the chunk size") + require.Contains(t, out.String(), "abandoned 2 of 2") +} + +func TestAbandonStuck_DryRunSendsNothing(t *testing.T) { + c := newStuckChain() + var out bytes.Buffer + n, err := AbandonStuck(context.Background(), AbandonStuckOpts{Client: c, Worker: me, LookbackBlocks: 5_000, DryRun: true, Out: &out}) + require.NoError(t, err) + require.Zero(t, n) + require.Empty(t, c.abandoned) + require.Contains(t, out.String(), "would abandon [2 5]") +} + +func TestAbandonStuck_OneRevertDoesNotStopTheSweep(t *testing.T) { + c := newStuckChain() + c.failOn = 2 + var out bytes.Buffer + n, err := AbandonStuck(context.Background(), AbandonStuckOpts{Client: c, Worker: me, LookbackBlocks: 5_000, Out: &out}) + require.Error(t, err) + require.Equal(t, 1, n) + require.Equal(t, []uint64{5}, c.abandoned) + require.Contains(t, err.Error(), "[2]") +} From 94c486fa86778e4a79093940510d96b823ee6754 Mon Sep 17 00:00:00 2001 From: Igor Date: Tue, 20 Oct 2026 10:27:44 -0500 Subject: [PATCH 2/2] abandon-stuck: check the Fprintf results errcheck flags; require JOB_REGISTRY_ADDRESS up front The four progress lines ignored fmt.Fprintf's result, which the CI lint (errcheck, new findings only) rejects. abandon-stuck also reads and writes the JobRegistry, so it takes loadAndValidateForSettlement like release does: a missing JOB_REGISTRY_ADDRESS fails at config load with a clear message instead of after the chain dial. --- cmd/cli/main.go | 2 +- internal/cli/abandon_stuck.go | 8 ++++---- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/cmd/cli/main.go b/cmd/cli/main.go index 68c39f9..03fb785 100644 --- a/cmd/cli/main.go +++ b/cmd/cli/main.go @@ -361,7 +361,7 @@ func runRemoveModels() { // runAbandonStuck sweeps this worker's in-flight backlog (see cli.AbandonStuck). // Each abandon is one transaction, so the budget is txTimeout per job plus the scan. func runAbandonStuck() { - cfg, logger := loadAndValidateCfg() + cfg, logger := loadAndValidateForSettlement() signingKey, workerAddr := loadSigningKey(cfg, logger) lookback := uint64(150_000) diff --git a/internal/cli/abandon_stuck.go b/internal/cli/abandon_stuck.go index c15fb3d..f58e08a 100644 --- a/internal/cli/abandon_stuck.go +++ b/internal/cli/abandon_stuck.go @@ -84,7 +84,7 @@ func AbandonStuck(ctx context.Context, o AbandonStuckOpts) (abandoned int, err e } } } - fmt.Fprintf(o.Out, "worker %s: %d jobs submitted in blocks %d-%d\n", o.Worker.Hex(), len(mine), from, head.Number) + _, _ = fmt.Fprintf(o.Out, "worker %s: %d jobs submitted in blocks %d-%d\n", o.Worker.Hex(), len(mine), from, head.Number) // Phase 2: the ones still in flight. var stuck []uint64 @@ -98,12 +98,12 @@ func AbandonStuck(ctx context.Context, o AbandonStuckOpts) (abandoned int, err e stuck = append(stuck, id) } } - fmt.Fprintf(o.Out, "in flight (Submitted/Acknowledged): %d\n", len(stuck)) + _, _ = fmt.Fprintf(o.Out, "in flight (Submitted/Acknowledged): %d\n", len(stuck)) if len(stuck) == 0 { return 0, nil } if o.DryRun { - fmt.Fprintf(o.Out, "dry run: would abandon %v\n", stuck) + _, _ = fmt.Fprintf(o.Out, "dry run: would abandon %v\n", stuck) return 0, nil } @@ -121,7 +121,7 @@ func AbandonStuck(ctx context.Context, o AbandonStuckOpts) (abandoned int, err e abandoned++ log.Info("job abandoned: fee refunded, in-flight slot released", "jobID", id) } - fmt.Fprintf(o.Out, "abandoned %d of %d\n", abandoned, len(stuck)) + _, _ = fmt.Fprintf(o.Out, "abandoned %d of %d\n", abandoned, len(stuck)) if len(failed) > 0 { return abandoned, fmt.Errorf("%d job(s) could not be abandoned (already settled, or the tx reverted): %v", len(failed), failed) }