diff --git a/internal/core/application/service.go b/internal/core/application/service.go index 7959fa386..6e1060200 100644 --- a/internal/core/application/service.go +++ b/internal/core/application/service.go @@ -3547,13 +3547,15 @@ func (s *service) scheduleSweepBatchOutput(round domain.Round) { return } - blockTimestamp, err := waitForConfirmation(context.Background(), round.CommitmentTxid, s.wallet) + // Use s.ctx so this poll stops on shutdown. On error we bail instead of + // guessing the height; the round is re-scheduled on next boot. + blockTimestamp, err := waitForConfirmation(s.ctx, round.CommitmentTxid, s.wallet) if err != nil { - log.WithError(err).Warnf( - "failed to wait for confirmation of commitment tx %s, schedule task time may be inaccurate", + log.WithError(err).Errorf( + "wallet unavailable; cannot schedule sweep for %s — will be picked up on next startup", round.CommitmentTxid, ) - blockTimestamp = &ports.BlockTimestamp{Time: time.Now().Unix()} + return } var expirationTimestamp int64 diff --git a/internal/core/application/sweeper.go b/internal/core/application/sweeper.go index 20e51ffb7..0d4786d54 100644 --- a/internal/core/application/sweeper.go +++ b/internal/core/application/sweeper.go @@ -470,15 +470,16 @@ func (s *sweeper) createBatchSweepTask(commitmentTxid, vtxoTreeRootTxid string) return } - // schedule AFTER the root input is confirmed + // Wait for the root input to confirm (s.ctx so it stops on shutdown). + // On error, bail instead of guessing the height; re-scheduled on next boot. rootInput := vtxoTree.Root.UnsignedTx.TxIn[0].PreviousOutPoint.Hash.String() - blockTimestamp, err := waitForConfirmation(context.Background(), rootInput, s.wallet) + blockTimestamp, err := waitForConfirmation(s.ctx, rootInput, s.wallet) if err != nil { - log.WithError(err).Warnf( - "failed to wait for confirmation of batch input tx %s, schedule task time "+ - "may be inaccurate", rootInput, + log.WithError(err).Errorf( + "wallet unavailable; cannot schedule sweep for batch input tx %s — "+ + "will be picked up on next startup", rootInput, ) - blockTimestamp = &ports.BlockTimestamp{Time: time.Now().Unix()} + return } var expirationTimestamp int64 diff --git a/internal/core/application/utils.go b/internal/core/application/utils.go index ba0cfbffb..e1a6cced9 100644 --- a/internal/core/application/utils.go +++ b/internal/core/application/utils.go @@ -541,7 +541,17 @@ func waitForConfirmation( return nil, ctx.Err() case <-ticker.C: confirmed, blockTimestamp, err := wallet.IsTransactionConfirmed(ctx, txid) - if confirmed && err == nil { + if err != nil { + // On shutdown, stop instead of retrying. + if ctx.Err() != nil { + return nil, ctx.Err() + } + log.WithError(err).Warnf( + "transient error checking confirmation of %s; will retry on next tick", txid, + ) + continue + } + if confirmed { log.Debugf( "tx %s confirmed at block height %d, block time %d", txid, @@ -550,9 +560,6 @@ func waitForConfirmation( ) return blockTimestamp, nil } - if err != nil { - return nil, err - } } } } diff --git a/internal/infrastructure/wallet/wallet_client.go b/internal/infrastructure/wallet/wallet_client.go index 1f4554197..e322d8c6b 100644 --- a/internal/infrastructure/wallet/wallet_client.go +++ b/internal/infrastructure/wallet/wallet_client.go @@ -5,11 +5,13 @@ import ( "encoding/hex" "fmt" "strings" + "time" "github.com/arkade-os/arkd/internal/core/domain" arklib "github.com/arkade-os/arkd/pkg/ark-lib" "github.com/btcsuite/btcd/btcec/v2" "github.com/btcsuite/btcd/wire" + grpc_retry "github.com/grpc-ecosystem/go-grpc-middleware/retry" log "github.com/sirupsen/logrus" arkwalletv1 "github.com/arkade-os/arkd/api-spec/protobuf/gen/arkwallet/v1" @@ -27,11 +29,25 @@ type walletDaemonClient struct { conn *grpc.ClientConn } +// retryCallOptions is the retry policy applied to every wallet gRPC call. +// It lives here (not inline in New) so tests exercise the same config. +func retryCallOptions() []grpc_retry.CallOption { + return []grpc_retry.CallOption{ + grpc_retry.WithMax(5), + grpc_retry.WithBackoff(grpc_retry.BackoffExponential(100 * time.Millisecond)), + grpc_retry.WithCodes(codes.Unavailable, codes.DeadlineExceeded, codes.ResourceExhausted), + } +} + // New creates a ports.WalletService backed by a gRPC client. func New(addr, otelCollectorEndpoint string) (ports.WalletService, *arklib.Network, error) { opts := []grpc.DialOption{ grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithUnaryInterceptor( + grpc_retry.UnaryClientInterceptor(retryCallOptions()...), + ), } + if otelCollectorEndpoint != "" { otelHandler := otelgrpc.NewClientHandler( otelgrpc.WithTracerProvider(otel.GetTracerProvider()), @@ -441,8 +457,12 @@ func (w *walletDaemonClient) GetCurrentBlockTime( func (w *walletDaemonClient) Withdraw( ctx context.Context, address string, amount uint64, all bool, ) (string, error) { - resp, err := w.client.Withdraw(ctx, &arkwalletv1.WithdrawRequest{ - Address: address, Amount: amount, All: all}, + // Don't retry Withdraw: it moves money, and a retry after an unclear + // failure could send it twice. WithMax(0) means zero retries. + resp, err := w.client.Withdraw( + ctx, + &arkwalletv1.WithdrawRequest{Address: address, Amount: amount, All: all}, + grpc_retry.WithMax(0), ) if err != nil { return "", err diff --git a/internal/infrastructure/wallet/wallet_client_test.go b/internal/infrastructure/wallet/wallet_client_test.go new file mode 100644 index 000000000..af97fdea4 --- /dev/null +++ b/internal/infrastructure/wallet/wallet_client_test.go @@ -0,0 +1,77 @@ +package walletclient + +import ( + "context" + "net" + "sync/atomic" + "testing" + + arkwalletv1 "github.com/arkade-os/arkd/api-spec/protobuf/gen/arkwallet/v1" + grpc_retry "github.com/grpc-ecosystem/go-grpc-middleware/retry" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" + "google.golang.org/grpc/test/bufconn" +) + +// fakeWalletServer counts Withdraw calls and always fails with a retryable error. +type fakeWalletServer struct { + arkwalletv1.UnimplementedWalletServiceServer + withdrawCalls atomic.Int32 +} + +func (s *fakeWalletServer) Withdraw( + _ context.Context, _ *arkwalletv1.WithdrawRequest, +) (*arkwalletv1.WithdrawResponse, error) { + s.withdrawCalls.Add(1) + return nil, status.Error(codes.Unavailable, "wallet unavailable") +} + +// TestWithdrawDoesNotRetry checks Withdraw runs at most once, so funds can't be sent twice. +func TestWithdrawDoesNotRetry(t *testing.T) { + srv := &fakeWalletServer{} + conn := newTestWalletConn(t, srv) + w := &walletDaemonClient{client: arkwalletv1.NewWalletServiceClient(conn), conn: conn} + + _, err := w.Withdraw(context.Background(), "bcrt1qaddr", 1000, false) + require.Error(t, err) + require.Equal(t, int32(1), srv.withdrawCalls.Load(), + "Withdraw must be attempted exactly once (double-spend risk)") + + // Control: without the WithMax(0) opt-out the same call IS retried, proving the interceptor is on. + srv.withdrawCalls.Store(0) + _, err = arkwalletv1.NewWalletServiceClient(conn).Withdraw( + context.Background(), + &arkwalletv1.WithdrawRequest{Address: "bcrt1qaddr", Amount: 1000}, + ) + require.Error(t, err) + require.Equal(t, int32(5), srv.withdrawCalls.Load(), + "control: a normal call should be retried 5 times by the interceptor") +} + +// newTestWalletConn returns an in-memory client conn using the same retry interceptor as New. +func newTestWalletConn(t *testing.T, srv arkwalletv1.WalletServiceServer) *grpc.ClientConn { + t.Helper() + + lis := bufconn.Listen(1 << 20) + grpcSrv := grpc.NewServer() + arkwalletv1.RegisterWalletServiceServer(grpcSrv, srv) + go func() { _ = grpcSrv.Serve(lis) }() + t.Cleanup(grpcSrv.Stop) + + conn, err := grpc.NewClient( + "passthrough:///bufconn", + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { + return lis.Dial() + }), + grpc.WithUnaryInterceptor( + grpc_retry.UnaryClientInterceptor(retryCallOptions()...), + ), + ) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + return conn +}