abandon-stuck: hand back the in-flight backlog with JobRegistry.abandonJob

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 := loadAndValidateForSettlement()
+	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)

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
+}

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]")
+}