Documentation
¶
Overview ¶
Package sweeps implements the data-model §7 sweep state machine over storage.Store with withdrawal-grade durability (FR-037–039):
- Exactly-once: the DB partial-unique constraint (at most one non-terminal SweepJob per (chain_id, source_address)) is honored — storage.ErrActiveSweepExists means a live sweep exists and NO second one is ever started, no matter how many balance snapshots arrive. The FR-039 idempotency key SWEEP:{chain_id}:{source}:{balance}:{height} is the subordinate request-dedup layer.
- Durability: sweeps take a sequences.Manager reservation via work_ref{kind: SWEEP}, persist the signed bytes at SIGNED, and follow the same unknown-outcome semantics as withdrawals — search first, rebroadcast the identical persisted bytes, quarantine the reservation on ambiguity. Never re-sign, never auto-release.
- Strategies (FR-038): FEE_RESERVE, FEE_FUND, THRESHOLD_ONLY, CUSTODY_ABSTRACTED. FEE_FUND emits the fee-wallet funding MsgSend through this same engine (its own SweepJob + reservation + persisted bytes); the scanner ledger-classifies that transfer FEE_FUNDING — internal, never a customer credit.
- Insufficient fee headroom defers the job (PENDING → DEFERRED): no retry loop — the deferral is counted (sovren_sweeps_deferred_total), logged, and revisited only when conditions change (Revisit). A deferred FEE_RESERVE/THRESHOLD_ONLY job whose stored amount is stale (the fee moved between plan and execution) is terminal-CANCELLED by Revisit once a fresh plan at current conditions passes, so the next Plan pass re-creates it with a recomputed amount — never a permanently stuck DEFERRED at a static balance.
Every economic threshold (minimum_sweep_amount_usovr, maximum_fee_percentage_for_sweep, fee_reserve_usovr) is configuration, never a code constant (FR-038/FR-040).
Index ¶
- Variables
- func FundingIdempotencyKey(chainID, feeWallet, parentSweepID string) string
- func IdempotencyKey(chainID, source string, balance sdkmath.Int, height uint64) string
- func IsFundingJob(j storage.SweepJob) bool
- func IsHold(err error) bool
- type Chain
- type Config
- type Engine
- func (e *Engine) Broadcast(ctx context.Context, sweepID string) error
- func (e *Engine) Confirm(ctx context.Context, sweepID string) error
- func (e *Engine) Pass(ctx context.Context) PassReport
- func (e *Engine) Plan(ctx context.Context) (PlanReport, error)
- func (e *Engine) Prepare(ctx context.Context, sweepID string) error
- func (e *Engine) Recover(ctx context.Context, sweepID string) error
- func (e *Engine) Revisit(ctx context.Context, sweepID string) (revived bool, err error)
- type Option
- type PassReport
- type PlanReport
- type SimulatePolicy
Constants ¶
This section is empty.
Variables ¶
var ( // ErrPaused reports an operational-controls pause for the sweep flow // (or the global signing/broadcast switches at those steps). ErrPaused = errors.New("sweeps: flow paused by operational controls") // ErrDeferred reports a sweep moved to DEFERRED by the current call: // fee headroom, fee percentage, or minimum-amount conditions failed. // The job is revisited when conditions change — never retried in a loop. ErrDeferred = errors.New("sweeps: sweep deferred (insufficient fee headroom or uneconomical)") // ErrFundingPending reports a FEE_FUND sweep waiting on its fee-wallet // funding transaction; the sweep stays PENDING until it confirms. ErrFundingPending = errors.New("sweeps: fee funding transfer in flight") // ErrNoFeeWallet reports the FEE_FUND strategy without an active // FEE_WALLET entry in the watch set — a configuration problem. ErrNoFeeWallet = errors.New("sweeps: FEE_FUND requires an active FEE_WALLET watched address") // policy: sweep processing holds (alert emitted), never guesses gas. ErrSimulationUnavailable = errors.New("sweeps: simulation unavailable; sweep held (simulate_unavailable: queue)") // at BUILT; nothing unsigned is ever broadcast. ErrSignerUnavailable = errors.New("sweeps: signer unavailable; sweep remains queued") // ErrQuarantined reports an ambiguous signer/broadcast outcome: the // sequence reservation is RECONCILIATION_REQUIRED and the sweep holds // its non-terminal status (still blocking any second live sweep) until // Recover rebroadcasts the persisted bytes or an operator resolves it. ErrQuarantined = errors.New("sweeps: quarantined; reservation requires reconciliation") // ErrNoSigner reports a transaction-emitting strategy configured // without a signer. ErrNoSigner = errors.New("sweeps: no signer configured") )
Functions ¶
func FundingIdempotencyKey ¶
FundingIdempotencyKey dedups the fee-wallet funding transfer to exactly one per parent sweep — a funding MsgSend can never be emitted twice for the same sweep, in any crash order.
func IdempotencyKey ¶
IdempotencyKey is the FR-039 sweep request-dedup key: SWEEP:{chain_id}:{source_address}:{balance_snapshot_usovr}:{snapshot_height}.
func IsFundingJob ¶
IsFundingJob reports whether j is a fee-wallet funding leg rather than a customer-address sweep.
Types ¶
type Chain ¶
type Chain interface {
Account(ctx context.Context, addr string) (accountNumber, sequence uint64, err error)
Balance(ctx context.Context, addr, denom string) (sdkmath.Int, error)
Simulate(ctx context.Context, txBytes []byte) (*client.SimulateResult, error)
Broadcast(ctx context.Context, txBytes []byte, mode client.BroadcastMode) (*client.BroadcastResult, error)
Tx(ctx context.Context, hash string) (*client.TxInfo, error)
NodeStatus(ctx context.Context) (*client.NodeStatus, error)
}
Chain is the subset of client.Client the engine needs; *client.Client implementations and the failover wrapper satisfy it.
type Config ¶
type Config struct {
ChainID string
// Strategy is the configured fee-handling strategy (FR-038).
Strategy storage.SweepStrategy
// HotWallet is the sweep destination (required unless
// CUSTODY_ABSTRACTED).
HotWallet string
// MinimumSweepAmountUsovr gates sweep creation and the final swept
// amount (integer base-unit string).
MinimumSweepAmountUsovr string
// MaximumFeePercentageForSweep is a decimal percentage string
// (e.g. "1.0"); a sweep whose fee exceeds amount × pct/100 defers.
MaximumFeePercentageForSweep string
// FeeReserveUsovr is the balance left behind under FEE_RESERVE
// (integer base-unit string).
FeeReserveUsovr string
// FeeWalletMaxSpendUsovr caps the fee wallet's cumulative FEE_FUND funding
// spend within FeeWalletSpendWindowBlocks recent blocks (integer base-unit
// string; "" or "0" disables the cap). A funding leg that would breach the
// cap DEFERS the sweep instead of spending — a guardrail so a dust flood or
// an address-derivation bug cannot drain the fee wallet one economical leg
// at a time (recommended for the FEE_FUND model; off by default).
FeeWalletMaxSpendUsovr string
// FeeWalletSpendWindowBlocks is the rolling block window the cap sums over;
// required (> 0) when FeeWalletMaxSpendUsovr is set.
FeeWalletSpendWindowBlocks uint64
// GasAdjustment is a decimal string applied to simulated gas with
// ceiling rounding; GasPriceUsovr is the decimal usovr-per-gas price.
GasAdjustment string
GasPriceUsovr string
// StaticGasLimit is required when SimulateUnavailable is SimulateStatic.
StaticGasLimit uint64
// BroadcastTimeout is the age after which a BROADCAST sweep whose
// transaction cannot be found enters the unknown flow.
BroadcastTimeout time.Duration
// Confirmations is the depth required for CONFIRMED.
Confirmations uint64
// KeyRefForSource maps a source address to the signer's opaque key
// handle. Nil means the source address itself is the key ref.
KeyRefForSource func(sourceAddress string) string
}
Config carries every economic value as a string; nothing here may be hard-coded in sweep logic (FR-038/FR-040).
type Engine ¶
type Engine struct {
// contains filtered or unexported fields
}
Engine drives SweepJobs through the §7 state machine. Safe for concurrent use: every status write goes through the repository layer, which enforces the legal transitions; concurrent drivers lose optimistically with ErrStatusConflict instead of double-executing, and the partial-unique constraint plus the sequence-reservation UNIQUE keys are the last-line exactly-once guarantees.
func New ¶
func New(store storage.Store, chain Chain, seq *sequences.Manager, sg signer.TransactionSigner, cfg Config, opts ...Option) (*Engine, error)
New validates cfg and builds an Engine. sg may be nil only under CUSTODY_ABSTRACTED (no transaction is ever emitted).
func (*Engine) Broadcast ¶
Broadcast pushes a SIGNED sweep's persisted bytes to the chain. It never signs: only j.SignedTxBytes ever reach the wire.
- CheckTx rejection ⇒ BROADCAST (code recorded) → FAILED; the reservation is quarantined (signed bytes exist and could still be accepted elsewhere — never auto-released). Covered deposits stay SWEEP_PENDING and re-attach to the next planned sweep.
- Acceptance ⇒ BROADCAST; inclusion and depth are Confirm's job.
- Transport error / timeout ⇒ the unknown flow: search for the original tx by hash first; found ⇒ treat as accepted; not found ⇒ the sweep HOLDS its SIGNED status (still the account's single live sweep) and the reservation is quarantined. Recovery is Recover — rebroadcasting the identical persisted bytes.
func (*Engine) Confirm ¶
Confirm advances a BROADCAST sweep from chain truth: DeliverTx failure ⇒ FAILED with the accurate code (the sequence WAS consumed and deposits stay SWEEP_PENDING for the next sweep); success at the configured depth ⇒ CONFIRMED, the reservation CONSUMED, and every covered deposit flipped SWEEP_PENDING → SWEPT in the same transaction. A sweep whose tx cannot be found after BroadcastTimeout has its reservation quarantined — never a re-sign.
func (*Engine) Pass ¶
func (e *Engine) Pass(ctx context.Context) PassReport
Pass runs one full sweeper iteration: plan new jobs, then advance every live job one step (PENDING/BUILT → Prepare, SIGNED → Broadcast, BROADCAST → Confirm, DEFERRED → Revisit). Hold-type conditions (pauses, funding in flight, signer/simulation outages, deferrals, lost optimistic races) are by-design waits, not errors.
func (*Engine) Plan ¶
func (e *Engine) Plan(ctx context.Context) (PlanReport, error)
Plan discovers sweepable sources and creates at most one SweepJob per source, honoring the non-terminal partial-unique constraint: a source with ANY live job (including DEFERRED and quarantined SIGNED/BROADCAST) is never given a second one, regardless of how many new balance snapshots have appeared (data model §7 guarantee 1).
Covered deposits (status CREDITED, plus SWEEP_PENDING orphans left by a terminal FAILED/CANCELLED sweep) flip CREDITED → SWEEP_PENDING in the same transaction that creates the job.
Under CUSTODY_ABSTRACTED no transaction is emitted: deposits settle by bookkeeping (CREDITED → SWEEP_PENDING → SWEPT) and funds stay put under the unified custody boundary.
func (*Engine) Prepare ¶
Prepare drives a PENDING (or crash-resumed BUILT) sweep to SIGNED: sequence reservation via work_ref{SWEEP}, final fee from simulation, strategy checks (defer on insufficiency — never a loop), the FEE_FUND funding gate, signing through the external signer boundary with adapter-side verification, and the atomic SIGNED persist (signed bytes + tx hash + reservation SIGNED in one transaction).
func (*Engine) Recover ¶
Recover drives a quarantined sweep (SIGNED or BROADCAST with a RECONCILIATION_REQUIRED reservation) by searching for and, only when absent, rebroadcasting the EXACT persisted bytes via the sequence manager. Re-signing is structurally impossible here. The reservation leaves quarantine only when Confirm observes the transaction on chain.
type Option ¶
type Option func(*Engine)
Option configures an Engine.
func WithMetrics ¶
WithMetrics attaches the adapter metric set (sweeps-deferred counter).
type PassReport ¶
type PassReport struct {
Plan PlanReport
Errors []error
}
PassReport summarizes one Pass.
type PlanReport ¶
type PlanReport struct {
// JobsCreated lists sweep IDs created in status PENDING.
JobsCreated []string
// JobsDeferred lists sweep IDs created and immediately DEFERRED (the
// snapshot cannot cover fee + reserve yet).
JobsDeferred []string
// CustodySwept counts deposits settled by CUSTODY_ABSTRACTED
// bookkeeping (no transaction).
CustodySwept int
// Held maps source address → reason for sources skipped this pass
// (active job, below minimum, simulation unavailable, ...).
Held map[string]string
}
PlanReport is the outcome of one Plan pass.
type SimulatePolicy ¶
type SimulatePolicy string
SimulatePolicy selects the behaviour when the node cannot serve Simulate (contracts/adapter-config-and-ops.md `simulate_unavailable`).
const ( // SimulateQueue holds sweeps and alerts (the safe default). SimulateQueue SimulatePolicy = "queue" // SimulateStatic uses the configured static MsgSend gas. Explicit opt-in. SimulateStatic SimulatePolicy = "static" )
Source Files