fix(sweeper): correct expiry handling and add the epoch sweep primitives

docs/version-compat.md

@@ -1,23 +1,24 @@
 # Build Version Compatibility
 
-The server enforces a minimum build (major) version globally, derived from its own
-build version. Any client whose major version is below the server's major
-version is rejected.
+The server enforces a minimum build version globally, derived from its own build
+version. Any client below that version is rejected.
 
-Only the **major** version component is compared. Minor and patch versions are
-ignored. For example, if the server is built as `v2.3.1`, a client at `2.0.0`,
-`2.1.0`, or `2.99.0` all pass, but `1.9.9` is rejected.
+The full `major.minor.patch` triple is compared, in that order. If the server is
+built as `v2.3.1` then `2.3.1` and `2.4.0` pass, but `2.3.0`, `2.2.9` and `1.9.9`
+are all rejected. Pre-release and build suffixes on the patch component are
+ignored, so `1.2.3-rc1` is treated as patch `3`.
 
 ## How it works
 
 1. At build time the server version is set via `-ldflags` (see
    `scripts/build-arkd`).
-2. On startup the server parses the major version from its build version string.
-3. On every request the server reads the client's SDK version header, extracts
-   the major version, and compares it against its own. If the client's major
-   version is lower, the request is rejected.
+2. On startup the server parses its own build version string.
+3. On every request the server reads the client's SDK version header and compares
+   it against its own, major first, then minor, then patch. If the client is
+   lower, the request is rejected.
 4. If a client does not send a header, the version check is skipped, allowing
-   backward compatibility.
+   backward compatibility — unless `build_version_header_required` is set, in
+   which case a missing or unparseable header is rejected.
 
 ### Client header
 
@@ -27,16 +28,23 @@ ignored. For example, if the server is built as `v2.3.1`, a client at `2.0.0`,
 | REST | `X-Build-Version` (HTTP header) |
 
 The value must be a semver string, optionally prefixed with `v` (e.g. `1.0.0` or
-`v1.0.0`). Only the major version component is used for comparison.
+`v1.0.0`). Missing components default to zero, so `1` is read as `1.0.0`.
 
 ### Decision table
 
 | Condition | Result |
 |-----------|--------|
-| No header sent | Request allowed |
-| Major version >= server major | Request allowed |
-| Major version < server major | `BUILD_VERSION_TOO_OLD` error |
-| Malformed version string | Request allowed |
+| No header sent | Allowed, unless `build_version_header_required` is set |
+| Malformed version string | Allowed, unless `build_version_header_required` is set |
+| Version >= server version | Request allowed |
+| Version < server version | `BUILD_VERSION_TOO_OLD` error |
+
+Two things worth knowing:
+
+- The guard applies only to the public `ArkService`. Admin and indexer RPCs are
+  never gated.
+- If the server's own build version cannot be parsed — a local or unreleased
+  build with no `-ldflags` — the guard is disabled for every client.
 
 ## Future: method-level versioning
 

internal/core/application/service.go

@@ -1828,14 +1828,15 @@ func (s *service) RegisterIntent(
 			continue
 		}
 
-		if settlementMinExpiryGap > 0 && !vtxo.Swept {
-			// reject if expires after now + settlementMinExpiryGap
-			expiresAt := time.Unix(vtxo.ExpiresAt, 0)
-			limit := time.Now().Add(settlementMinExpiryGap)
-			if expiresAt.After(limit) {
+		// A swept vtxo is exempt: settling one is how recovery works, and the
+		// operator already holds the funds onchain.
+		if !vtxo.Swept {
+			if err := checkSettlementExpiryGap(
+				time.Unix(vtxo.ExpiresAt, 0), time.Now(), settlementMinExpiryGap,
+			); err != nil {
 				return "", errors.INVALID_PSBT_INPUT.New(
-					"vtxo %s expires after %s (minExpiryGap: %s)",
-					vtxo.Outpoint.String(), limit, settlementMinExpiryGap,
+					"vtxo %s: %s (minExpiryGap: %s)",
+					vtxo.Outpoint.String(), err, settlementMinExpiryGap,
 				).WithMetadata(errors.InputMetadata{
 					Txid:       proofTxid,
 					InputIndex: int(outpoint.Index),
@@ -3168,12 +3169,7 @@ func (s *service) startFinalization(
 	flatVtxoTree := make(tree.FlatTxTree, 0)
 	if vtxoTree != nil {
 
-		sweepClosure := script.CSVMultisigClosure{
-			MultisigClosure: script.MultisigClosure{PubKeys: []*btcec.PublicKey{forfeitPubkey}},
-			Locktime:        vtxoTreeExpiry,
-		}
-
-		sweepScript, err := sweepClosure.Script()
+		root, _, err := tree.BuildLegacySweepTapTreeRoot(forfeitPubkey, vtxoTreeExpiry)
 		if err != nil {
 			return
 		}
@@ -3184,10 +3180,6 @@ func (s *service) startFinalization(
 		}
 		batchOutputAmount := commitmentPtx.UnsignedTx.TxOut[0].Value
 
-		sweepLeaf := txscript.NewBaseTapLeaf(sweepScript)
-		sweepTapTree := txscript.AssembleTaprootScriptTree(sweepLeaf)
-		root := sweepTapTree.RootNode.TapHash()
-
 		coordinator, err := tree.NewTreeCoordinatorSession(
 			root.CloneBytes(), batchOutputAmount, vtxoTree,
 		)

internal/core/application/sweeper.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"encoding/hex"
-	"errors"
 	"fmt"
 	"strings"
 	"sync"
@@ -383,7 +382,11 @@ func (s *sweeper) scheduleTask(task sweeperTask) error {
 			"sweeper: trying to schedule task in the past for tx %s, executing it immediately",
 			task.id,
 		)
-		return task.execute()
+		// Same retry policy as the scheduled path. This is the branch a restart
+		// takes for a batch that expired while the service was down, so it is the
+		// one that most needs retrying: nothing re-arms the task until the next
+		// restart.
+		return s.executeWithRetry(task)
 	}
 
 	s.locker.Lock()
@@ -408,14 +411,67 @@ func (s *sweeper) scheduleTask(task sweeperTask) error {
 		}
 		s.locker.Unlock()
 
+		// Release the dedup slot before running: scheduledTasks guards against
+		// registering the same task twice, so holding the id past execution would
+		// block every later schedule for this tree, including our own retries.
 		s.removeTask(task.id)
 
-		if err := task.execute(); err != nil {
-			log.WithError(err).Errorf("failed to execute sweep of tx %s", task.id)
-		}
+		// Nothing to return the error to from a scheduler callback; executeWithRetry
+		// has already logged the give-up.
+		_ = s.executeWithRetry(task)
 	})
 }
 
+var (
+	// sweepRetryDelay is how long to wait before re-attempting a sweep whose
+	// execution failed. A failure is usually a mempool conflict or a transient
+	// node error, neither of which clears in milliseconds.
+	sweepRetryDelay = time.Minute
+	// sweepRetryAttempts bounds the in-process retries. Beyond this the sweep is
+	// left for the next process start, which rebuilds tasks from the repository.
+	sweepRetryAttempts = 10
+)
+
+// executeWithRetry runs a sweep task, re-attempting it on failure. Without this a
+// single failed broadcast left the batch outputs unswept until an operator
+// restart, which is the window an attacker racing the sweep needs.
+// Returns the last error if every attempt failed, so callers running a task
+// inline still learn it did not succeed.
+func (s *sweeper) executeWithRetry(task sweeperTask) error {
+	ctx := s.ctx
+	if ctx == nil {
+		ctx = context.Background()
+	}
+
+	err := task.execute()
+	if err == nil {
+		return nil
+	}
+
+	for attempt := 1; attempt <= sweepRetryAttempts; attempt++ {
+		log.WithError(err).Warnf(
+			"sweeper: sweep of tx %s failed, retrying in %s (attempt %d/%d)",
+			task.id, sweepRetryDelay, attempt, sweepRetryAttempts,
+		)
+
+		select {
+		case <-ctx.Done():
+			return err
+		case <-time.After(sweepRetryDelay):
+		}
+
+		if err = task.execute(); err == nil {
+			return nil
+		}
+	}
+
+	log.WithError(err).Errorf(
+		"sweeper: giving up on sweep of tx %s after %d attempts, outputs remain unswept "+
+			"until the next restart", task.id, sweepRetryAttempts,
+	)
+	return err
+}
+
 // createBatchSweepTask returns a function passed as handler in the scheduler
 // it tries to craft a sweep tx containing the onchain outputs of the given vtxo tree
 // if some parts of the tree have been broadcasted in the meantime, it will schedule the next
@@ -651,8 +707,10 @@ func (s *sweeper) createBatchSweepTask(commitmentTxid, vtxoTreeRootTxid string)
 			}
 
 			err = nil
-			// retry until the tx is broadcasted or the error is not BIP68 final
-			for len(txid) == 0 && (err == nil || errors.Is(err, ports.ErrNonFinalBIP68)) {
+			// retry until the tx is broadcasted or the error is not a timelock
+			// one. Both kinds matter: batch outputs can be gated by a relative
+			// sequence, an absolute nLockTime, or both.
+			for len(txid) == 0 && (err == nil || ports.IsNonFinal(err)) {
 				select {
 				case <-s.ctx.Done():
 					return nil

internal/core/application/sweeper_test.go

@@ -685,3 +685,163 @@ func newTestSweeper(t *testing.T) (
 	s := newSweeper(wallet, repoManager, builder, scheduler)
 	return wallet, vtxoRepo, markerRepo, builder, s
 }
+
+// controllableScheduler captures the scheduled closure so a test can fire it on
+// demand. mockScheduler above is a no-op and cannot drive scheduleTask.
+type controllableScheduler struct {
+	fn func()
+}
+
+func (s *controllableScheduler) Start()               {}
+func (s *controllableScheduler) Stop()                {}
+func (s *controllableScheduler) Unit() ports.TimeUnit { return ports.UnixTime }
+func (s *controllableScheduler) AfterNow(expiry int64) bool {
+	return true // always "in the future" so scheduleTask registers rather than running inline
+}
+func (s *controllableScheduler) ScheduleTaskOnce(at int64, task func()) error {
+	s.fn = task
+	return nil
+}
+func (s *controllableScheduler) fire() {
+	if s.fn != nil {
+		s.fn()
+	}
+}
+
+// newSchedulableSweeper builds a sweeper wired to a scheduler the test controls.
+// newTestSweeper above uses mockScheduler, whose ScheduleTaskOnce is a no-op.
+func newSchedulableSweeper(sched ports.SchedulerService) *sweeper {
+	s := newSweeper(&mockWalletService{}, &mockRepoManager{}, &mockTxBuilder{}, sched)
+	s.ctx = context.Background()
+	return s
+}
+
+// TestScheduleTaskRetriesFailedExecution pins that a sweep whose broadcast fails
+// is re-attempted rather than silently dropped. Before this, a single mempool
+// conflict left the batch unswept until an operator restart.
+func TestScheduleTaskRetriesFailedExecution(t *testing.T) {
+	prevDelay, prevAttempts := sweepRetryDelay, sweepRetryAttempts
+	sweepRetryDelay, sweepRetryAttempts = time.Millisecond, 3
+	t.Cleanup(func() { sweepRetryDelay, sweepRetryAttempts = prevDelay, prevAttempts })
+
+	sched := &controllableScheduler{}
+	s := newSchedulableSweeper(sched)
+
+	calls := 0
+	require.NoError(t, s.scheduleTask(sweeperTask{
+		id: "task-fail",
+		at: time.Now().Add(time.Hour).Unix(),
+		execute: func() error {
+			calls++
+			return fmt.Errorf("broadcast rejected")
+		},
+	}))
+
+	sched.fire()
+
+	// one initial attempt plus sweepRetryAttempts retries
+	require.Equal(t, 1+3, calls, "a failed sweep must be retried, not dropped")
+}
+
+// TestScheduleTaskStopsRetryingOnSuccess pins that a retry that succeeds ends the
+// loop instead of burning the remaining attempts.
+func TestScheduleTaskStopsRetryingOnSuccess(t *testing.T) {
+	prevDelay, prevAttempts := sweepRetryDelay, sweepRetryAttempts
+	sweepRetryDelay, sweepRetryAttempts = time.Millisecond, 5
+	t.Cleanup(func() { sweepRetryDelay, sweepRetryAttempts = prevDelay, prevAttempts })
+
+	sched := &controllableScheduler{}
+	s := newSchedulableSweeper(sched)
+
+	calls := 0
+	require.NoError(t, s.scheduleTask(sweeperTask{
+		id: "task-eventually-ok",
+		at: time.Now().Add(time.Hour).Unix(),
+		execute: func() error {
+			calls++
+			if calls < 3 {
+				return fmt.Errorf("still conflicting")
+			}
+			return nil
+		},
+	}))
+
+	sched.fire()
+	require.Equal(t, 3, calls, "retrying must stop as soon as the sweep succeeds")
+}
+
+// TestScheduleTaskFreesDedupSlot pins that the id is released once the task has
+// run, so a later schedule for the same tree is accepted. scheduledTasks is a
+// dedup guard: holding the id past execution would block re-scheduling entirely.
+func TestScheduleTaskFreesDedupSlot(t *testing.T) {
+	prevDelay, prevAttempts := sweepRetryDelay, sweepRetryAttempts
+	sweepRetryDelay, sweepRetryAttempts = time.Millisecond, 1
+	t.Cleanup(func() { sweepRetryDelay, sweepRetryAttempts = prevDelay, prevAttempts })
+
+	sched := &controllableScheduler{}
+	s := newSchedulableSweeper(sched)
+
+	at := time.Now().Add(time.Hour).Unix()
+	failing := sweeperTask{
+		id: "task-slot", at: at,
+		execute: func() error { return fmt.Errorf("broadcast rejected") },
+	}
+	require.NoError(t, s.scheduleTask(failing))
+
+	s.locker.Lock()
+	_, registered := s.scheduledTasks["task-slot"]
+	s.locker.Unlock()
+	require.True(t, registered, "task must occupy the dedup slot while pending")
+
+	sched.fire()
+
+	s.locker.Lock()
+	_, stillHeld := s.scheduledTasks["task-slot"]
+	s.locker.Unlock()
+	require.False(t, stillHeld, "dedup slot must be released so the id can be re-scheduled")
+
+	second := 0
+	require.NoError(t, s.scheduleTask(sweeperTask{
+		id: "task-slot", at: at,
+		execute: func() error { second++; return nil },
+	}))
+	sched.fire()
+	require.Equal(t, 1, second, "a fresh schedule for the same id must be accepted")
+}
+
+// TestScheduleTaskRetriesAlreadyDueTask covers the path a restart takes. When a
+// batch expired while the service was down, AfterNow is false and scheduleTask
+// runs the task inline rather than handing it to the scheduler. That path must
+// use the same bounded retry policy - it is the case that most needs it, since
+// nothing will re-arm the task until the next restart.
+func TestScheduleTaskRetriesAlreadyDueTask(t *testing.T) {
+	prevDelay, prevAttempts := sweepRetryDelay, sweepRetryAttempts
+	sweepRetryDelay, sweepRetryAttempts = time.Millisecond, 3
+	t.Cleanup(func() { sweepRetryDelay, sweepRetryAttempts = prevDelay, prevAttempts })
+
+	sched := &immediateScheduler{}
+	s := newSchedulableSweeper(sched)
+
+	calls := 0
+	err := s.scheduleTask(sweeperTask{
+		id: "already-due",
+		at: time.Now().Add(-time.Hour).Unix(),
+		execute: func() error {
+			calls++
+			return fmt.Errorf("broadcast rejected")
+		},
+	})
+
+	require.Error(t, err, "the final failure must still reach the caller")
+	require.Equal(t, 1+3, calls, "an already-due task must retry like a scheduled one")
+}
+
+// immediateScheduler reports every task as already due, which is what a restart
+// looks like for a batch that expired while the service was down.
+type immediateScheduler struct{}
+
+func (s *immediateScheduler) Start()                               {}
+func (s *immediateScheduler) Stop()                                {}
+func (s *immediateScheduler) Unit() ports.TimeUnit                 { return ports.UnixTime }
+func (s *immediateScheduler) AfterNow(expiry int64) bool           { return false }
+func (s *immediateScheduler) ScheduleTaskOnce(int64, func()) error { return nil }