diff --git a/internal/evm/abi.go b/internal/evm/abi.go index 2e221ff..f01c7b4 100644 --- a/internal/evm/abi.go +++ b/internal/evm/abi.go @@ -14,6 +14,10 @@ import ( // Contract ABIs for Autonomi storage payments. // Sourced from evmlib's IPaymentVault ABI. +// MerklePoolCandidateCount is the contract's fixed candidate count per pool +// (the CandidateNode[16] tuple below). +const MerklePoolCandidateCount = 16 + var payForQuotesABI abi.ABI var payForMerkleTreeABI abi.ABI var merklePaymentMadeEvent abi.Event diff --git a/internal/evm/signer.go b/internal/evm/signer.go index 47afb4d..bcd9ab6 100644 --- a/internal/evm/signer.go +++ b/internal/evm/signer.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "math/big" + "sort" "strings" "sync" "time" @@ -281,8 +282,8 @@ func (s *Signer) PayForMerkleTree( } commitments[i].PoolHash = poolHash - if len(pc.Candidates) != 16 { - return "", "", fmt.Errorf("pool %d: expected 16 candidates, got %d", i, len(pc.Candidates)) + if len(pc.Candidates) != MerklePoolCandidateCount { + return "", "", fmt.Errorf("pool %d: expected %d candidates, got %d", i, MerklePoolCandidateCount, len(pc.Candidates)) } for j, c := range pc.Candidates { amt, ok := new(big.Int).SetString(c.Amount, 10) @@ -296,11 +297,12 @@ func (s *Signer) PayForMerkleTree( } } - // Ensure token allowance for the merkle payments contract. We don't know the - // exact cost upfront (the contract picks one winning candidate per pool), so - // approve the maximum it could possibly charge rather than an unlimited - // max-uint256 allowance. - if err := s.ensureAllowance(ctx, privateKey, fromAddress, tokenAddr, merkleAddr, maxMerklePayout(commitments)); err != nil { + // Ensure token allowance for the merkle payments contract. We don't know + // the exact cost upfront (the contract charges the winner pool's median + // quote * 2^depth, and picks the winner at execution time), so approve the + // maximum it could possibly charge rather than an unlimited max-uint256 + // allowance. + if err := s.ensureAllowance(ctx, privateKey, fromAddress, tokenAddr, merkleAddr, maxMerklePayout(depth, commitments)); err != nil { return "", "", fmt.Errorf("token approval: %w", err) } @@ -412,22 +414,122 @@ func (s *Signer) sendTxWithReceipt( return signedTx.Hash().Hex(), receipt, nil } -// maxMerklePayout returns the largest total the merkle payment contract could -// charge: the sum over pools of the highest candidate amount in each pool (the -// contract pays exactly one winning candidate per pool). Used to bound the -// ERC-20 allowance instead of approving max-uint256. -func maxMerklePayout(commitments []MerklePoolCommitment) *big.Int { +// merkleMaxDepth mirrors the contract's MAX_MERKLE_DEPTH; used only to clamp +// the estimation shift below (real depth validation happens in the worker and +// on-chain). +const merkleMaxDepth = 8 + +// sortedMedian returns the charging median of a pool's candidate amounts: the +// element at index len/2 of the sorted list — for the contract's fixed 16 +// candidates that is index 8, exactly median16's pick in PaymentVaultV2. +func sortedMedian(amounts []*big.Int) *big.Int { + if len(amounts) == 0 { + return new(big.Int) + } + sort.Slice(amounts, func(i, j int) bool { return amounts[i].Cmp(amounts[j]) < 0 }) + return amounts[len(amounts)/2] +} + +// merkleTreeCharge returns the most one payForMerkleTree call could charge. +// PaymentVaultV2 charges median16(winner pool quotes) * 2^depth, and the +// winner pool is not known until the transaction executes, so the bound takes +// the highest pool median. Unparseable amounts count as zero — strict +// validation happens in the worker before any money moves. +func merkleTreeCharge(depth int, pools []antd.PoolCommitmentEntry) *big.Int { + maxMedian := new(big.Int) + for _, pc := range pools { + amounts := make([]*big.Int, 0, len(pc.Candidates)) + for _, c := range pc.Candidates { + amt, ok := new(big.Int).SetString(c.Amount, 10) + if !ok { + amt = new(big.Int) + } + amounts = append(amounts, amt) + } + if m := sortedMedian(amounts); m.Cmp(maxMedian) > 0 { + maxMedian = m + } + } + if depth < 0 { + depth = 0 + } else if depth > merkleMaxDepth { + depth = merkleMaxDepth + } + return new(big.Int).Lsh(maxMedian, uint(depth)) +} + +// MaxMerkleBatchesPayout returns the largest total a merkle payment plan could +// charge: the sum over batches of that batch's worst-case tree charge. Shared +// by the worker's max_gas_fee precheck and EnsureMerkleAllowance so the two +// always agree. +func MaxMerkleBatchesPayout(batches []antd.MerkleBatchEntry) *big.Int { total := new(big.Int) + for _, b := range batches { + total.Add(total, merkleTreeCharge(b.Depth, b.PoolCommitments)) + } + return total +} + +// EnsureMerkleAllowance approves the ERC-20 allowance for a whole multi-batch +// merkle payment plan in one transaction: the sum of every batch's maximum +// payout. Each subsequent PayForMerkleTree call re-checks the allowance and +// finds it sufficient (spend per batch never exceeds that batch's maximum), so +// the per-batch approval short-circuits — one approve tx instead of N. +func (s *Signer) EnsureMerkleAllowance( + ctx context.Context, + privateKeyHex string, + batches []antd.MerkleBatchEntry, + tokenAddress string, + merklePaymentsAddress string, +) error { + required := MaxMerkleBatchesPayout(batches) + if required.Sign() == 0 { + return nil + } + + s.mu.Lock() + defer s.mu.Unlock() + + privateKeyHex = strings.TrimPrefix(privateKeyHex, "0x") + privateKey, err := crypto.HexToECDSA(privateKeyHex) + if err != nil { + return fmt.Errorf("invalid private key: %w", err) + } + fromAddress := crypto.PubkeyToAddress(privateKey.PublicKey) + + if err := s.ensureAllowance(ctx, privateKey, fromAddress, + common.HexToAddress(tokenAddress), common.HexToAddress(merklePaymentsAddress), required); err != nil { + return fmt.Errorf("token approval: %w", err) + } + return nil +} + +// maxMerklePayout returns the largest total one payForMerkleTree call could +// charge: PaymentVaultV2 charges median16(winner pool quotes) * 2^depth, and +// the winner pool is unknown until the transaction executes, so the bound +// takes the highest pool median. Used to bound the ERC-20 allowance instead +// of approving max-uint256. +func maxMerklePayout(depth int, commitments []MerklePoolCommitment) *big.Int { + maxMedian := new(big.Int) for _, pc := range commitments { - poolMax := new(big.Int) + amounts := make([]*big.Int, 0, len(pc.Candidates)) for _, c := range pc.Candidates { - if c.Amount != nil && c.Amount.Cmp(poolMax) > 0 { - poolMax = c.Amount + amt := c.Amount + if amt == nil { + amt = new(big.Int) } + amounts = append(amounts, amt) + } + if m := sortedMedian(amounts); m.Cmp(maxMedian) > 0 { + maxMedian = m } - total.Add(total, poolMax) } - return total + if depth < 0 { + depth = 0 + } else if depth > merkleMaxDepth { + depth = merkleMaxDepth + } + return new(big.Int).Lsh(maxMedian, uint(depth)) } // waitForReceipt polls for a transaction receipt with a 2-second interval. diff --git a/internal/evm/signer_test.go b/internal/evm/signer_test.go index 41561d3..9a3968c 100644 --- a/internal/evm/signer_test.go +++ b/internal/evm/signer_test.go @@ -3,36 +3,100 @@ package evm import ( "math/big" "testing" + + antd "github.com/WithAutonomi/ant-sdk/antd-go" ) +// mkPool builds a full 16-candidate ABI pool. Fewer than 16 amounts are +// padded with nils (counted as zero quotes), mirroring how a partially +// populated pool would price. +func mkPool(amounts ...int64) MerklePoolCommitment { + var pc MerklePoolCommitment + for i, a := range amounts { + pc.Candidates[i] = MerkleCandidateNode{Amount: big.NewInt(a)} + } + return pc +} + +func full16(v int64) []int64 { + out := make([]int64, 16) + for i := range out { + out[i] = v + } + return out +} + func TestMaxMerklePayout(t *testing.T) { - mk := func(amounts ...int64) MerklePoolCommitment { - var pc MerklePoolCommitment - for i, a := range amounts { - pc.Candidates[i] = MerkleCandidateNode{Amount: big.NewInt(a)} - } - // remaining candidates stay zero-valued (Amount == nil) - return pc + // PaymentVaultV2 charges median16(winner pool) * 2^depth; the bound takes + // the highest pool median. + oneToSixteen := make([]int64, 16) + for i := range oneToSixteen { + oneToSixteen[i] = int64(i + 1) } tests := []struct { name string + depth int commitments []MerklePoolCommitment want int64 }{ - {"empty", nil, 0}, - {"single pool, max is last", []MerklePoolCommitment{mk(10, 50, 30)}, 50}, - {"single pool, max is first", []MerklePoolCommitment{mk(99, 1, 2)}, 99}, - {"two pools sum of maxes", []MerklePoolCommitment{mk(10, 50), mk(7, 3, 100)}, 150}, - {"pool with all nil candidates", []MerklePoolCommitment{{}}, 0}, + {"empty", 4, nil, 0}, + {"uniform pool: median times 2^depth", 3, []MerklePoolCommitment{mkPool(full16(10)...)}, 80}, + {"median16 is the upper median (index 8)", 2, []MerklePoolCommitment{mkPool(oneToSixteen...)}, 36}, // sorted[8] = 9, << 2 + {"highest pool median wins", 4, []MerklePoolCommitment{mkPool(full16(5)...), mkPool(full16(7)...)}, 112}, + // 8 explicit amounts + 8 nil-padded zeros: the zeros sort first, so + // sorted index 8 is the smallest real quote (10); depth 0 leaves it + // unshifted. + {"nil amounts count as zero quotes", 0, []MerklePoolCommitment{mkPool(10, 20, 30, 40, 50, 60, 70, 80)}, 10}, + {"depth clamped to contract max 8", 20, []MerklePoolCommitment{mkPool(full16(1)...)}, 256}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - got := maxMerklePayout(tt.commitments) + got := maxMerklePayout(tt.depth, tt.commitments) if got.Cmp(big.NewInt(tt.want)) != 0 { t.Errorf("maxMerklePayout = %s, want %d", got, tt.want) } }) } } + +func TestMaxMerkleBatchesPayout(t *testing.T) { + pool := func(amounts ...string) antd.PoolCommitmentEntry { + var pc antd.PoolCommitmentEntry + for _, a := range amounts { + pc.Candidates = append(pc.Candidates, antd.CandidateNodeEntry{Amount: a}) + } + return pc + } + + tests := []struct { + name string + batches []antd.MerkleBatchEntry + want string + }{ + {"empty", nil, "0"}, + {"single batch: max pool median shifted by depth", []antd.MerkleBatchEntry{ + {Depth: 3, PoolCommitments: []antd.PoolCommitmentEntry{ + pool("10", "70", "30"), // sorted [10 30 70], median idx 1 = 30 + pool("5", "9"), // median idx 1 = 9 + }}, + }, "240"}, // 30 << 3 + {"multi batch sums per-batch charges", []antd.MerkleBatchEntry{ + {Depth: 3, PoolCommitments: []antd.PoolCommitmentEntry{pool("10", "70", "30"), pool("5", "9")}}, // 240 + {Depth: 1, PoolCommitments: []antd.PoolCommitmentEntry{pool("100")}}, // 100 << 1 + }, "440"}, + {"unparseable amounts count as zero quotes", []antd.MerkleBatchEntry{ + {Depth: 1, PoolCommitments: []antd.PoolCommitmentEntry{pool("not-a-number", "40")}}, // sorted [0 40] median idx 1 = 40 + }, "80"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := MaxMerkleBatchesPayout(tt.batches) + if got.String() != tt.want { + t.Errorf("MaxMerkleBatchesPayout = %s, want %s", got, tt.want) + } + }) + } +} diff --git a/internal/worker/upload.go b/internal/worker/upload.go index f5b42cc..cc13454 100644 --- a/internal/worker/upload.go +++ b/internal/worker/upload.go @@ -84,22 +84,20 @@ func classifyFailure(err error, paymentMade, canRetry bool) uploadOutcome { } // estimatedUploadCost returns the cost to compare against max_gas_fee. For -// wave-batch it's the quoted TotalAmount; for merkle (cost determined on-chain) -// it's the upper bound the contract could charge — the sum over pools of the -// largest candidate amount. Unparseable/empty amounts count as zero. +// wave-batch it's the quoted TotalAmount; for merkle (cost determined +// on-chain) it's the upper bound the contract could charge across every +// payment batch — per batch, the highest pool median quote * 2^depth +// (PaymentVaultV2 charges the winner pool's median16 * 2^depth and picks the +// winner at execution time). Unparseable/empty amounts count as zero; a +// merkle prepare with no batches at all estimates zero here and fails with a +// clear error in the payment branch. func estimatedUploadCost(prepared *antd.PrepareUploadResult) *big.Int { if prepared.PaymentType == "merkle" { - total := new(big.Int) - for _, pc := range prepared.PoolCommitments { - poolMax := new(big.Int) - for _, c := range pc.Candidates { - if amt, ok := new(big.Int).SetString(c.Amount, 10); ok && amt.Cmp(poolMax) > 0 { - poolMax = amt - } - } - total.Add(total, poolMax) + batches, err := merkleBatchPlan(prepared) + if err != nil { + return new(big.Int) } - return total + return evm.MaxMerkleBatchesPayout(batches) } if amt, ok := new(big.Int).SetString(strings.TrimSpace(prepared.TotalAmount), 10); ok { return amt @@ -107,6 +105,90 @@ func estimatedUploadCost(prepared *antd.PrepareUploadResult) *big.Int { return new(big.Int) } +// merkleBatchPlan returns the ordered merkle payment batches for a prepare +// result. antd >= 0.12.0 always populates MerkleBatches (one entry per +// payForMerkleTree transaction the signer must submit); older daemons fill +// only the legacy single-batch mirror fields, synthesized here into one batch. +// Neither present means the daemon and SDK disagree about the wire format — +// surfaced as a clear version-gap error instead of an empty payment attempt. +func merkleBatchPlan(prepared *antd.PrepareUploadResult) ([]antd.MerkleBatchEntry, error) { + if len(prepared.MerkleBatches) > 0 { + return prepared.MerkleBatches, nil + } + if len(prepared.PoolCommitments) > 0 { + return []antd.MerkleBatchEntry{{ + Depth: prepared.Depth, + PoolCommitments: prepared.PoolCommitments, + MerklePaymentTimestamp: prepared.MerklePaymentTimestamp, + }}, nil + } + return nil, fmt.Errorf("merkle prepare returned no payment batches and no pool commitments — daemon/SDK version mismatch (multi-batch merkle uploads need antd >= 0.12.0)") +} + +// validateMerkleBatches rejects malformed payment data before any money moves, +// so a bad batch N can never fail after batches 1..N-1 already paid. Mirrors +// the checks PayForMerkleTree applies per batch, naming the batch and pool. +func validateMerkleBatches(batches []antd.MerkleBatchEntry) error { + for i, b := range batches { + if b.Depth < 1 || b.Depth > 8 { + return fmt.Errorf("batch %d/%d has invalid merkle depth %d (want 1-8)", i+1, len(batches), b.Depth) + } + if len(b.PoolCommitments) == 0 { + return fmt.Errorf("batch %d/%d has no pool commitments", i+1, len(batches)) + } + for j, pc := range b.PoolCommitments { + if len(pc.Candidates) != evm.MerklePoolCandidateCount { + return fmt.Errorf("batch %d/%d pool %d: expected %d candidates, got %d", + i+1, len(batches), j, evm.MerklePoolCandidateCount, len(pc.Candidates)) + } + for _, c := range pc.Candidates { + if _, ok := new(big.Int).SetString(c.Amount, 10); !ok { + return fmt.Errorf("batch %d/%d pool %d: invalid candidate amount %q", + i+1, len(batches), j, c.Amount) + } + } + } + } + return nil +} + +// padWinnerList right-pads winners with "" to batchCount entries. The daemon +// requires the finalize list length to equal the batch count; an empty entry +// marks a batch the signer never paid. +func padWinnerList(winners []string, batchCount int) []string { + padded := make([]string, batchCount) + copy(padded, winners) + return padded +} + +// shouldSalvageMerkle decides whether a mid-plan payment failure warrants a +// best-effort partial finalize (storing the already-paid batches' chunks). +// Only when something was actually paid, and never on a confirmation timeout — +// that tx may still mine, and a partial finalize now would consume the upload +// and foreclose the manual full finalize. +func shouldSalvageMerkle(err error, paidBatches int) bool { + return paidBatches > 0 && !errors.Is(err, evm.ErrConfirmationTimeout) +} + +// logIfPartialUpload logs the chunk accounting when err carries antd's +// PARTIAL_UPLOAD detail (some batches/chunks stored, the rest failed). +func logIfPartialUpload(uuid string, err error) { + var partial *antd.PartialUploadError + if errors.As(err, &partial) { + slog.Warn("merkle finalize stored only part of the upload", + "uuid", uuid, "chunks_stored", partial.ChunksStored, + "chunks_failed", partial.ChunksFailed, "total_chunks", partial.TotalChunks) + } +} + +// merkleAllowancePayer is the optional pre-approval seam: implemented by +// *evm.Signer. Payers without it (e.g. a hosted-mode gateway payer) fall back +// to PayForMerkleTree's own per-batch approval — pre-approval only saves the +// N-1 extra approve transactions. +type merkleAllowancePayer interface { + EnsureMerkleAllowance(ctx context.Context, privateKeyHex string, batches []antd.MerkleBatchEntry, tokenAddress, merklePaymentsAddress string) error +} + // UploadWorker processes queued file uploads in the background. type UploadWorker struct { uploadSvc *services.UploadService @@ -468,33 +550,62 @@ func (w *UploadWorker) processUpload(ctx context.Context, upload *services.Uploa switch prepared.PaymentType { case "merkle": - // Phase 2: Sign merkle batch payment - winnerHash, totalPaid, err := w.evmSigner.PayForMerkleTree( - ctx, walletKey, - prepared.Depth, - prepared.PoolCommitments, - prepared.MerklePaymentTimestamp, - tokenAddr, - prepared.PaymentVaultAddress, - ) - if err != nil { - return fmt.Errorf("EVM merkle payment failed: %w", err) + // Phase 2: Sign merkle payments — one payForMerkleTree transaction per + // batch (antd >= 0.12.0 splits uploads larger than one merkle tree, 256 + // fresh chunks ~= 1 GiB, into several batches; older daemons send a + // single batch via the legacy mirror fields). + batches, planErr := merkleBatchPlan(prepared) + if planErr != nil { + return planErr // nothing paid → abandon + } + if verr := validateMerkleBatches(batches); verr != nil { + return fmt.Errorf("merkle payment data invalid: %w", verr) // nothing paid → abandon + } + slog.Info("merkle payment plan", "uuid", upload.UUID, "batches", len(batches), + "est_max_cost", evm.MaxMerkleBatchesPayout(batches).String()) + + // Multi-batch: pre-approve the whole plan's worst case in one approve + // tx. Optional seam — payers without it self-approve per batch. + if len(batches) > 1 { + if p, ok := any(w.evmSigner).(merkleAllowancePayer); ok { + if aerr := p.EnsureMerkleAllowance(ctx, walletKey, batches, tokenAddr, prepared.PaymentVaultAddress); aerr != nil { + return fmt.Errorf("EVM merkle allowance approval failed: %w", aerr) + } + } } - paidAmount = totalPaid - txHash = winnerHash - - slog.Info("EVM merkle payment submitted", - "uuid", upload.UUID, "winner_pool_hash", winnerHash, "total_paid", totalPaid) - // Record the confirmed spend BEFORE finalize, so a finalize failure still - // leaves an accounting record rather than losing the payment (V2-426). - w.recordPayment(ctx, wallet, upload, tokenAddr, paidAmount, txHash) + winners, totalPaid, payErr := w.payMerkleBatches(ctx, walletKey, batches, tokenAddr, + prepared.PaymentVaultAddress, wallet, upload) + if payErr != nil { + if shouldSalvageMerkle(payErr, len(winners)) { + // Definitive failure with money already spent: store what was + // paid for so a retry pays only the remainder, then preserve. + if r := w.salvageMerkleFinalize(ctx, upload, prepared.UploadID, winners, len(batches)); r != nil { + result = r + paidAmount = totalPaid.String() + break // salvage fully succeeded — complete the upload + } + return fmt.Errorf("EVM merkle payment failed at batch %d/%d (%d paid): %w", + len(winners)+1, len(batches), len(winners), errors.Join(errPaidNoRetry, payErr)) + } + // Nothing paid, or a confirmation timeout (the tx may still mine — + // no salvage, so the manual full finalize stays possible). %w keeps + // evm.ErrConfirmationTimeout visible to classifyFailure. + return fmt.Errorf("EVM merkle payment failed (batch %d/%d): %w", + len(winners)+1, len(batches), payErr) + } + paidAmount = totalPaid.String() // Phase 3: Finalize merkle upload. A failure here means money is already // spent; re-running would submit a second merkle payment (not provably // zero-cost), so flag it no-retry and preserve the source for recovery. - result, err = w.antdClient.FinalizeMerkleUpload(ctx, prepared.UploadID, winnerHash, false) + if len(batches) == 1 { + result, err = w.antdClient.FinalizeMerkleUpload(ctx, prepared.UploadID, winners[0], false) + } else { + result, err = w.antdClient.FinalizeMerkleUploadMulti(ctx, prepared.UploadID, winners, false) + } if err != nil { + logIfPartialUpload(upload.UUID, err) return fmt.Errorf("Failed to finalize merkle upload: %w", errors.Join(errPaidNoRetry, err)) } @@ -604,6 +715,83 @@ func (w *UploadWorker) recordPayment(ctx context.Context, wallet *services.Walle } } +// payMerkleBatches submits one payForMerkleTree transaction per batch, in +// order. Each confirmed batch is recorded (one transactions row per batch, +// tx_hash = that batch's winner pool hash) BEFORE the next batch is paid +// (V2-426), so a mid-plan failure leaves an accurate accounting trail. Returns +// the winner hashes of the batches paid so far — on error, len(winners) is the +// paid count and the error is the failing batch's raw error (caller wraps). +func (w *UploadWorker) payMerkleBatches( + ctx context.Context, + walletKey string, + batches []antd.MerkleBatchEntry, + tokenAddr, vaultAddr string, + wallet *services.Wallet, + upload *services.Upload, +) (winners []string, totalPaid *big.Int, err error) { + winners = make([]string, 0, len(batches)) + totalPaid = new(big.Int) + for i, b := range batches { + slog.Info("paying merkle batch", "uuid", upload.UUID, + "batch", i+1, "batches", len(batches), "pools", len(b.PoolCommitments)) + winnerHash, batchPaid, payErr := w.evmSigner.PayForMerkleTree( + ctx, walletKey, b.Depth, b.PoolCommitments, b.MerklePaymentTimestamp, tokenAddr, vaultAddr) + if payErr != nil { + return winners, totalPaid, payErr + } + winners = append(winners, winnerHash) + if amt, ok := new(big.Int).SetString(batchPaid, 10); ok { + totalPaid.Add(totalPaid, amt) + } + w.recordPayment(ctx, wallet, upload, tokenAddr, batchPaid, winnerHash) + slog.Info("merkle batch paid", "uuid", upload.UUID, + "batch", i+1, "batches", len(batches), "winner_pool_hash", winnerHash, + "amount", batchPaid, "cumulative", totalPaid.String()) + } + return winners, totalPaid, nil +} + +// salvageMerkleFinalize is the best-effort finalize after a definitive +// mid-plan payment failure: the paid batches' chunks still store (unpaid +// batches surface through PARTIAL_UPLOAD), so a later re-upload of the same +// content dedups them and pays only for the remainder. Runs detached from the +// caller's cancellation (bounded) so a worker shutdown that killed the payment +// doesn't also skip the salvage. Returns a non-nil result only in the +// unexpected full-success case (the unpaid batches' chunks were already +// on-network) — the upload is then genuinely complete and the caller should +// continue the normal completion path. +func (w *UploadWorker) salvageMerkleFinalize( + ctx context.Context, + upload *services.Upload, + uploadID string, + winners []string, + batchCount int, +) *antd.FinalizeUploadResult { + slog.Warn("merkle payment failed mid-plan — salvage-finalizing paid batches", + "uuid", upload.UUID, "paid", len(winners), "batches", batchCount) + + salvageCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Minute) + defer cancel() + + result, err := w.antdClient.FinalizeMerkleUploadMulti(salvageCtx, uploadID, padWinnerList(winners, batchCount), false) + if err == nil { + slog.Warn("merkle salvage finalize fully succeeded — unpaid batches were already on-network", + "uuid", upload.UUID, "chunks", result.ChunksStored) + return result + } + + var partial *antd.PartialUploadError + if errors.As(err, &partial) { + slog.Warn("merkle salvage finalize stored paid batches", + "uuid", upload.UUID, "chunks_stored", partial.ChunksStored, + "chunks_failed", partial.ChunksFailed, "total_chunks", partial.TotalChunks) + } else { + slog.Warn("merkle salvage finalize failed — paid batches remain pending on antd (a rejected finalize does not consume the upload; manual retry possible)", + "uuid", upload.UUID, "upload_id", uploadID, "error", err) + } + return nil +} + // seedDownloadCache write-through-seeds the download cache from the upload's // temp file (V2-822): the bytes are already staged on the cache's own volume, // so admission is a rename — zero extra I/O — and the deferred temp-file diff --git a/internal/worker/upload_worker_test.go b/internal/worker/upload_worker_test.go index d298d4e..9307857 100644 --- a/internal/worker/upload_worker_test.go +++ b/internal/worker/upload_worker_test.go @@ -5,6 +5,7 @@ import ( "fmt" "os" "path/filepath" + "strings" "testing" "time" @@ -376,6 +377,12 @@ func TestClassifyFailure(t *testing.T) { merklePaid := errors.Join(errPaidNoRetry, errors.New("finalize reverted")) permanent := &antd.BadRequestError{AntdError: antd.AntdError{StatusCode: 400, Message: "bad"}} generic := errors.New("boom") + // The exact wrapping shapes the multi-batch merkle branch produces (V2-1056): + merkleMidPlanPaid := fmt.Errorf("EVM merkle payment failed at batch 3/5 (2 paid): %w", + errors.Join(errPaidNoRetry, errors.New("transaction reverted"))) + merkleMidPlanTimeout := fmt.Errorf("EVM merkle payment failed (batch 3/5): %w", evm.ErrConfirmationTimeout) + merklePartialFinalize := errors.Join(errPaidNoRetry, &antd.PartialUploadError{ + AntdError: antd.AntdError{StatusCode: 502, Message: "partial"}, ChunksStored: 300, ChunksFailed: 12, TotalChunks: 312}) tests := []struct { name string @@ -395,6 +402,9 @@ func TestClassifyFailure(t *testing.T) { {"permanent error, nothing paid → abandon immediately", permanent, false, true, outcomeAbandon}, {"generic error after payment → preserve", generic, true, true, outcomePreservePaid}, {"generic error, nothing paid → abandon", generic, false, true, outcomeAbandon}, + {"merkle mid-plan definitive failure after paid batches → preserve paid", merkleMidPlanPaid, true, true, outcomePreservePaid}, + {"merkle mid-plan confirmation timeout → preserve unconfirmed", merkleMidPlanTimeout, true, true, outcomePreserveUnconfirmed}, + {"merkle multi finalize partial → preserve paid (not transient-retried)", merklePartialFinalize, true, true, outcomePreservePaid}, } for _, tt := range tests { @@ -423,16 +433,202 @@ func TestEstimatedUploadCost_WaveUnparseable(t *testing.T) { } func TestEstimatedUploadCost_Merkle(t *testing.T) { - // Two pools; cost ceiling = sum of the max candidate amount in each pool. + // Legacy single-batch shape. PaymentVaultV2 charges the winner pool's + // median quote * 2^depth; the ceiling takes the highest pool median. p := &antd.PrepareUploadResult{ PaymentType: "merkle", + Depth: 3, PoolCommitments: []antd.PoolCommitmentEntry{ - {Candidates: []antd.CandidateNodeEntry{{Amount: "10"}, {Amount: "70"}, {Amount: "30"}}}, - {Candidates: []antd.CandidateNodeEntry{{Amount: "5"}, {Amount: "9"}}}, + {Candidates: []antd.CandidateNodeEntry{{Amount: "10"}, {Amount: "70"}, {Amount: "30"}}}, // median 30 + {Candidates: []antd.CandidateNodeEntry{{Amount: "5"}, {Amount: "9"}}}, // median 9 }, } - if got := estimatedUploadCost(p); got.String() != "79" { // 70 + 9 - t.Errorf("merkle cost = %s, want 79", got) + if got := estimatedUploadCost(p); got.String() != "240" { // 30 << 3 + t.Errorf("merkle cost = %s, want 240", got) + } +} + +func TestEstimatedUploadCost_MerkleMultiBatch(t *testing.T) { + // Ceiling sums per-batch charges across ALL payment batches, not just the + // legacy mirror. + p := &antd.PrepareUploadResult{ + PaymentType: "merkle", + MerkleBatches: []antd.MerkleBatchEntry{ + {Depth: 2, PoolCommitments: []antd.PoolCommitmentEntry{ + {Candidates: []antd.CandidateNodeEntry{{Amount: "10"}, {Amount: "70"}}}, // median 70 + }}, + {Depth: 1, PoolCommitments: []antd.PoolCommitmentEntry{ + {Candidates: []antd.CandidateNodeEntry{{Amount: "5"}, {Amount: "9"}}}, // median 9 + {Candidates: []antd.CandidateNodeEntry{{Amount: "100"}}}, // median 100 + }}, + }, + } + if got := estimatedUploadCost(p); got.String() != "480" { // (70 << 2) + (100 << 1) + t.Errorf("multi-batch merkle cost = %s, want 480", got) + } +} + +func TestEstimatedUploadCost_MerkleEmptyBothIsZero(t *testing.T) { + // No batches and no legacy fields: estimate is zero — the precheck passes + // and the payment branch then fails with the clear version-gap error. + p := &antd.PrepareUploadResult{PaymentType: "merkle"} + if got := estimatedUploadCost(p); got.Sign() != 0 { + t.Errorf("empty merkle cost = %s, want 0", got) + } +} + +// --- multi-batch merkle helpers (V2-1056) --- + +func fullPool(amount string) antd.PoolCommitmentEntry { + pc := antd.PoolCommitmentEntry{PoolHash: "0x01"} + for i := 0; i < evm.MerklePoolCandidateCount; i++ { + pc.Candidates = append(pc.Candidates, antd.CandidateNodeEntry{Amount: amount}) + } + return pc +} + +func TestMerkleBatchPlan_UsesBatches(t *testing.T) { + batches := []antd.MerkleBatchEntry{ + {Depth: 8, MerklePaymentTimestamp: 111}, + {Depth: 6, MerklePaymentTimestamp: 222}, + {Depth: 4, MerklePaymentTimestamp: 333}, + } + got, err := merkleBatchPlan(&antd.PrepareUploadResult{PaymentType: "merkle", MerkleBatches: batches}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(got) != 3 || got[0].MerklePaymentTimestamp != 111 || got[2].MerklePaymentTimestamp != 333 { + t.Errorf("plan = %+v, want the 3 batches in order", got) + } +} + +func TestMerkleBatchPlan_LegacyFallback(t *testing.T) { + p := &antd.PrepareUploadResult{ + PaymentType: "merkle", + Depth: 5, + PoolCommitments: []antd.PoolCommitmentEntry{fullPool("10")}, + MerklePaymentTimestamp: 999, + } + got, err := merkleBatchPlan(p) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(got) != 1 || got[0].Depth != 5 || got[0].MerklePaymentTimestamp != 999 || len(got[0].PoolCommitments) != 1 { + t.Errorf("plan = %+v, want one batch synthesized from the legacy fields", got) + } +} + +func TestMerkleBatchPlan_PrefersBatchesOverLegacyMirror(t *testing.T) { + // antd >= 0.12.0 single-batch shape: MerkleBatches[0] AND the legacy mirror. + p := &antd.PrepareUploadResult{ + PaymentType: "merkle", + Depth: 5, + PoolCommitments: []antd.PoolCommitmentEntry{fullPool("10")}, + MerklePaymentTimestamp: 999, + MerkleBatches: []antd.MerkleBatchEntry{{Depth: 5, MerklePaymentTimestamp: 999}}, + } + got, err := merkleBatchPlan(p) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(got) != 1 || len(got[0].PoolCommitments) != 0 { + t.Errorf("plan = %+v, want MerkleBatches verbatim (not a legacy synthesis)", got) + } +} + +func TestMerkleBatchPlan_EmptyBothVersionGapError(t *testing.T) { + _, err := merkleBatchPlan(&antd.PrepareUploadResult{PaymentType: "merkle"}) + if err == nil { + t.Fatal("want error for merkle prepare with no batches and no legacy fields") + } + if !strings.Contains(err.Error(), "antd >= 0.12.0") { + t.Errorf("error %q should name the version gap", err) + } +} + +func TestValidateMerkleBatches(t *testing.T) { + shortPool := antd.PoolCommitmentEntry{Candidates: make([]antd.CandidateNodeEntry, 15)} + for i := range shortPool.Candidates { + shortPool.Candidates[i].Amount = "1" + } + badAmountPool := fullPool("1") + badAmountPool.Candidates[3].Amount = "not-a-number" + + tests := []struct { + name string + batches []antd.MerkleBatchEntry + wantErr string // empty = valid + }{ + {"valid multi-batch", []antd.MerkleBatchEntry{ + {Depth: 8, PoolCommitments: []antd.PoolCommitmentEntry{fullPool("1"), fullPool("2")}}, + {Depth: 5, PoolCommitments: []antd.PoolCommitmentEntry{fullPool("3")}}, + }, ""}, + {"invalid depth", []antd.MerkleBatchEntry{ + {Depth: 9, PoolCommitments: []antd.PoolCommitmentEntry{fullPool("1")}}, + }, "batch 1/1 has invalid merkle depth 9 (want 1-8)"}, + {"zero depth", []antd.MerkleBatchEntry{ + {PoolCommitments: []antd.PoolCommitmentEntry{fullPool("1")}}, + }, "batch 1/1 has invalid merkle depth 0 (want 1-8)"}, + {"batch without pools", []antd.MerkleBatchEntry{ + {Depth: 8, PoolCommitments: []antd.PoolCommitmentEntry{fullPool("1")}}, + {Depth: 5}, + }, "batch 2/2 has no pool commitments"}, + {"wrong candidate count names batch and pool", []antd.MerkleBatchEntry{ + {Depth: 8, PoolCommitments: []antd.PoolCommitmentEntry{fullPool("1")}}, + {Depth: 5, PoolCommitments: []antd.PoolCommitmentEntry{fullPool("1"), shortPool}}, + }, "batch 2/2 pool 1: expected 16 candidates, got 15"}, + {"unparseable amount names batch and pool", []antd.MerkleBatchEntry{ + {Depth: 8, PoolCommitments: []antd.PoolCommitmentEntry{badAmountPool}}, + }, `batch 1/1 pool 0: invalid candidate amount "not-a-number"`}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := validateMerkleBatches(tt.batches) + if tt.wantErr == "" { + if err != nil { + t.Errorf("unexpected error: %v", err) + } + return + } + if err == nil || err.Error() != tt.wantErr { + t.Errorf("error = %v, want %q", err, tt.wantErr) + } + }) + } +} + +func TestPadWinnerList(t *testing.T) { + got := padWinnerList([]string{"0xa", "0xb"}, 5) + want := []string{"0xa", "0xb", "", "", ""} + if len(got) != len(want) { + t.Fatalf("len = %d, want %d", len(got), len(want)) + } + for i := range want { + if got[i] != want[i] { + t.Errorf("padded[%d] = %q, want %q", i, got[i], want[i]) + } + } + if full := padWinnerList([]string{"0xa"}, 1); len(full) != 1 || full[0] != "0xa" { + t.Errorf("already-full list changed: %v", full) + } + if empty := padWinnerList(nil, 2); len(empty) != 2 || empty[0] != "" || empty[1] != "" { + t.Errorf("nil winners = %v, want two empty entries", empty) + } +} + +func TestShouldSalvageMerkle(t *testing.T) { + definitive := errors.New("transaction reverted: 0xdead") + timeout := fmt.Errorf("payment: %w", evm.ErrConfirmationTimeout) + + if !shouldSalvageMerkle(definitive, 2) { + t.Error("definitive failure with paid batches should salvage") + } + if shouldSalvageMerkle(definitive, 0) { + t.Error("nothing paid → nothing to salvage") + } + if shouldSalvageMerkle(timeout, 2) { + t.Error("confirmation timeout must NOT salvage — the tx may still mine") } }