From dc4f97286b568d6f00016b1d58c41c0b05da4344 Mon Sep 17 00:00:00 2001 From: Daniil Yankouski Date: Mon, 6 Jul 2026 05:11:10 +0200 Subject: [PATCH] fix(devshard): record delivered-but-unsettled requests as delivered_pending_settlement, not failed (#1387) --- devshard/cmd/devshardctl/escrow_rotator.go | 80 ++++++++++-- .../escrow_settlement_retry_test.go | 91 ++++++++++++++ devshard/cmd/devshardctl/gateway.go | 2 + devshard/cmd/devshardctl/redundancy.go | 58 +++++++-- .../redundancy_delivered_pending_test.go | 115 ++++++++++++++++++ devshard/types/errors.go | 1 - devshard/user/session.go | 41 +++---- devshard/user/warmkey_test.go | 55 ++++++++- 8 files changed, 396 insertions(+), 47 deletions(-) create mode 100644 devshard/cmd/devshardctl/escrow_settlement_retry_test.go create mode 100644 devshard/cmd/devshardctl/redundancy_delivered_pending_test.go diff --git a/devshard/cmd/devshardctl/escrow_rotator.go b/devshard/cmd/devshardctl/escrow_rotator.go index 1300a419ab..d6d478f74e 100644 --- a/devshard/cmd/devshardctl/escrow_rotator.go +++ b/devshard/cmd/devshardctl/escrow_rotator.go @@ -109,6 +109,7 @@ func (g *Gateway) rotateEscrowsOnce() { } if !pocActive { g.finishBridgeEscrows(snapshot, settings) + g.retrySettlements(context.Background(), snapshot.EpochIndex, settings) } } @@ -151,6 +152,7 @@ func (g *Gateway) prepareBridgeEscrows(snapshot ChainPhaseSnapshot, settings Gat if err != nil { log.Printf("escrow_rotation_regular_retire_failed epoch=%d model=%q escrow=%s error=%v", epoch, model.ModelID, devshard.ID, err) settleFailed++ + g.registerSettlementRetry(devshard.ID, epoch+1) } else if settledOnChain { settled++ } @@ -219,6 +221,7 @@ func (g *Gateway) finishBridgeEscrows(snapshot ChainPhaseSnapshot, settings Gate if err != nil { log.Printf("escrow_rotation_temp_retire_failed epoch=%d model=%q escrow=%s error=%v", epoch, model.ModelID, devshard.ID, err) settleFailed++ + g.registerSettlementRetry(devshard.ID, epoch+1) } else if settledOnChain { settled++ } @@ -411,19 +414,16 @@ func (g *Gateway) settleDevshardOnChain(ctx context.Context, id string, req admi return nil, err } log.Printf("devshard_settle_key_loaded escrow=%s settler=%s key_env=%q", id, signer.Address(), privateKeyEnv) - if rt.proxy.sm.Phase() != types.PhaseSettlement { - g.finalizeMu.Lock() - log.Printf("gateway_finalize_lock_acquired escrow=%s path=rotation_settle", id) - if err := rt.session.Finalize(ctx); err != nil { - g.finalizeMu.Unlock() - log.Printf("devshard_settle_failed escrow=%s stage=finalize error=%q", id, err.Error()) - return nil, err - } - g.finalizeMu.Unlock() - log.Printf("devshard_settle_finalize_completed escrow=%s phase=%s", id, sessionPhaseLabel(rt.proxy.sm.Phase())) - } else { - log.Printf("devshard_settle_finalize_skipped escrow=%s phase=%s", id, sessionPhaseLabel(rt.proxy.sm.Phase())) - } + g.finalizeMu.Lock() + log.Printf("gateway_finalize_lock_acquired escrow=%s path=rotation_settle", id) + // Always finalize: the PhaseSettlement branch re-collects a short quorum, so a retry after a validator returns completes instead of re-broadcasting the same short proof. + finalizeErr := rt.session.Finalize(ctx) + g.finalizeMu.Unlock() + if finalizeErr != nil { + log.Printf("devshard_settle_failed escrow=%s stage=finalize error=%q", id, finalizeErr.Error()) + return nil, finalizeErr + } + log.Printf("devshard_settle_finalize_completed escrow=%s phase=%s", id, sessionPhaseLabel(rt.proxy.sm.Phase())) settlement, err := rt.proxy.settlementJSON() if err != nil { log.Printf("devshard_settle_failed escrow=%s stage=settlement_json error=%q", id, err.Error()) @@ -445,5 +445,59 @@ func (g *Gateway) settleDevshardOnChain(ctx context.Context, id string, req admi return nil, err } log.Printf("devshard_settle_submitted escrow=%s tx_hash=%s settler=%s", id, result.TxHash, result.Settler) + g.clearSettlementRetry(id) return result, nil } + +// registerSettlementRetry queues a devshard whose settlement failed for retry through deadlineEpoch (inclusive), keeping the earliest deadline if already queued. +func (g *Gateway) registerSettlementRetry(id string, deadlineEpoch uint64) { + g.settlementRetryMu.Lock() + defer g.settlementRetryMu.Unlock() + if g.settlementRetry == nil { + g.settlementRetry = make(map[string]uint64) + } + if _, exists := g.settlementRetry[id]; !exists { + g.settlementRetry[id] = deadlineEpoch + } +} + +func (g *Gateway) clearSettlementRetry(id string) { + g.settlementRetryMu.Lock() + defer g.settlementRetryMu.Unlock() + delete(g.settlementRetry, id) +} + +func (g *Gateway) pendingSettlementRetries() map[string]uint64 { + g.settlementRetryMu.Lock() + defer g.settlementRetryMu.Unlock() + if len(g.settlementRetry) == 0 { + return nil + } + out := make(map[string]uint64, len(g.settlementRetry)) + for id, deadline := range g.settlementRetry { + out[id] = deadline + } + return out +} + +// retrySettlements re-attempts failed settlements until success or the deadline epoch; past that the on-chain reaper takes over. Success clears the entry inside settleDevshardOnChain. +func (g *Gateway) retrySettlements(ctx context.Context, currentEpoch uint64, settings GatewaySettings) { + if !settings.EscrowRotation.SettlementEnabled { + return + } + for id, deadlineEpoch := range g.pendingSettlementRetries() { + if currentEpoch > deadlineEpoch { + g.clearSettlementRetry(id) + log.Printf("settlement_retry_expired escrow=%s epoch=%d deadline=%d", id, currentEpoch, deadlineEpoch) + continue + } + attemptCtx, cancel := context.WithTimeout(ctx, autoSettlementAttemptTimeout) + _, err := gatewaySettleDevshardOnChain(g, attemptCtx, id, adminSettleEscrowRequest{}) + cancel() + if err != nil { + log.Printf("settlement_retry_failed escrow=%s epoch=%d deadline=%d error=%v", id, currentEpoch, deadlineEpoch, err) + continue + } + log.Printf("settlement_retry_succeeded escrow=%s epoch=%d", id, currentEpoch) + } +} diff --git a/devshard/cmd/devshardctl/escrow_settlement_retry_test.go b/devshard/cmd/devshardctl/escrow_settlement_retry_test.go new file mode 100644 index 0000000000..cf6ac1151e --- /dev/null +++ b/devshard/cmd/devshardctl/escrow_settlement_retry_test.go @@ -0,0 +1,91 @@ +package main + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/require" +) + +func settlementEnabledSettings() GatewaySettings { + return GatewaySettings{EscrowRotation: EscrowRotationSettings{SettlementEnabled: true}} +} + +func stubSettleDevshardOnChain(t *testing.T, fn func(id string) (*SettleDevshardEscrowResult, error)) { + t.Helper() + saved := gatewaySettleDevshardOnChain + t.Cleanup(func() { gatewaySettleDevshardOnChain = saved }) + gatewaySettleDevshardOnChain = func(g *Gateway, _ context.Context, id string, _ adminSettleEscrowRequest) (*SettleDevshardEscrowResult, error) { + result, err := fn(id) + if err == nil { + g.clearSettlementRetry(id) // mirror the real settleDevshardOnChain, which clears on success + } + return result, err + } +} + +func TestSettlementRetry_RetriesUntilSuccessThenClears(t *testing.T) { + attempts := 0 + stubSettleDevshardOnChain(t, func(string) (*SettleDevshardEscrowResult, error) { + attempts++ + if attempts < 3 { + return nil, errors.New("insufficient signatures") + } + return &SettleDevshardEscrowResult{}, nil + }) + + gateway := &Gateway{} + gateway.registerSettlementRetry("escrow-1", 5) + settings := settlementEnabledSettings() + + gateway.retrySettlements(context.Background(), 5, settings) + require.Contains(t, gateway.pendingSettlementRetries(), "escrow-1", "still failing, kept for retry") + + gateway.retrySettlements(context.Background(), 5, settings) + require.Contains(t, gateway.pendingSettlementRetries(), "escrow-1") + + gateway.retrySettlements(context.Background(), 5, settings) + require.NotContains(t, gateway.pendingSettlementRetries(), "escrow-1", "cleared once settled") + require.Equal(t, 3, attempts) +} + +func TestSettlementRetry_ExpiresPastDeadlineWithoutRetrying(t *testing.T) { + attempts := 0 + stubSettleDevshardOnChain(t, func(string) (*SettleDevshardEscrowResult, error) { + attempts++ + return nil, errors.New("insufficient signatures") + }) + + gateway := &Gateway{} + gateway.registerSettlementRetry("escrow-1", 5) + + gateway.retrySettlements(context.Background(), 6, settlementEnabledSettings()) + + require.NotContains(t, gateway.pendingSettlementRetries(), "escrow-1", "dropped past deadline") + require.Equal(t, 0, attempts, "no on-chain call past the deadline") +} + +func TestSettlementRetry_NoopWhenSettlementDisabled(t *testing.T) { + attempts := 0 + stubSettleDevshardOnChain(t, func(string) (*SettleDevshardEscrowResult, error) { + attempts++ + return &SettleDevshardEscrowResult{}, nil + }) + + gateway := &Gateway{} + gateway.registerSettlementRetry("escrow-1", 5) + + gateway.retrySettlements(context.Background(), 5, GatewaySettings{}) + + require.Contains(t, gateway.pendingSettlementRetries(), "escrow-1", "left untouched while settlement disabled") + require.Equal(t, 0, attempts) +} + +func TestRegisterSettlementRetry_KeepsEarliestDeadline(t *testing.T) { + gateway := &Gateway{} + gateway.registerSettlementRetry("escrow-1", 5) + gateway.registerSettlementRetry("escrow-1", 9) + + require.Equal(t, uint64(5), gateway.pendingSettlementRetries()["escrow-1"]) +} diff --git a/devshard/cmd/devshardctl/gateway.go b/devshard/cmd/devshardctl/gateway.go index d1c62eca91..733f7b84da 100644 --- a/devshard/cmd/devshardctl/gateway.go +++ b/devshard/cmd/devshardctl/gateway.go @@ -62,6 +62,8 @@ type Gateway struct { settlementInFlight map[string]struct{} replenishmentMu sync.Mutex replenishmentInFlight map[string]struct{} + settlementRetryMu sync.Mutex + settlementRetry map[string]uint64 // devshard id -> last epoch to retry on-chain settlement (inclusive) mu sync.Mutex roundRobinSeed atomic.Uint64 } diff --git a/devshard/cmd/devshardctl/redundancy.go b/devshard/cmd/devshardctl/redundancy.go index 7e7c7430e2..324f183fcb 100644 --- a/devshard/cmd/devshardctl/redundancy.go +++ b/devshard/cmd/devshardctl/redundancy.go @@ -3304,12 +3304,14 @@ func (e *Redundancy) finishRaceOutcome(ctx context.Context, attempts []*inflight e.recordGatewayTimeoutAction(inf, params, timeoutResultKind(result, inf), "completed", "none") } } + // Only relabels accounting/metrics; escrow and the returned error stay untouched. + terminal := e.resolveTerminalDecision(attempts, winnerNonce, failed) if hostErr := hostApplicationErrorFromAttempts(attempts, winnerNonce); hostErr != nil { captureAllAttemptsFailedRequest(ctx, e.devshardID, params, hostErr) - logRequestStage(ctx, "request_failed", "escrow", e.devshardID, "error", hostErr) - e.recordGatewayRequestOutcome(params.Model, "failed", gatewayRequestFailureReason(failed)) - e.completeAccountingRequest(ctx, 0, decision, "failed") - e.logRequestSettled(ctx, 0, decision, "failed") + logRequestStage(ctx, terminal.stage, "escrow", e.devshardID, "error", hostErr) + e.recordGatewayRequestOutcome(params.Model, terminal.outcome, terminal.reason) + e.completeAccountingRequest(ctx, terminal.nonce, decision, terminal.outcome) + e.logRequestSettled(ctx, terminal.nonce, decision, terminal.outcome) e.checkEscrowMissing(ctx, attempts) return hostErr } @@ -3321,10 +3323,10 @@ func (e *Redundancy) finishRaceOutcome(ctx context.Context, attempts []*inflight errMsg = (&nonStreamingReducedMaxTokensTimeoutError{}).Error() } captureAllAttemptsFailedRequest(ctx, e.devshardID, params, fmt.Errorf("%s", errMsg)) - logRequestStage(ctx, "request_failed", "escrow", e.devshardID, "error", errMsg) - e.recordGatewayRequestOutcome(params.Model, "failed", gatewayRequestFailureReason(failed)) - e.completeAccountingRequest(ctx, 0, decision, "failed") - e.logRequestSettled(ctx, 0, decision, "failed") + logRequestStage(ctx, terminal.stage, "escrow", e.devshardID, "error", errMsg) + e.recordGatewayRequestOutcome(params.Model, terminal.outcome, terminal.reason) + e.completeAccountingRequest(ctx, terminal.nonce, decision, terminal.outcome) + e.logRequestSettled(ctx, terminal.nonce, decision, terminal.outcome) e.checkEscrowMissing(ctx, attempts) if opts.nonStreamingReducedTokenTimeout { return &nonStreamingReducedMaxTokensTimeoutError{} @@ -3463,6 +3465,46 @@ func (e *Redundancy) resolvedWinnerNonce(attempts []*inflight, winnerNonce uint6 return 0 } +// winnerDeliveredPendingSettlement reports whether the crowned winner streamed clean content to the client but its nonce never finished — the executor delivered a response yet produced no applied finish, so it cannot settle. Not a failure from the client's view. +func (e *Redundancy) winnerDeliveredPendingSettlement(attempts []*inflight, winnerNonce uint64) bool { + if winnerNonce == 0 { + return false + } + winner := inflightByNonce(attempts, winnerNonce) + if winner == nil || winner.probe { + return false + } + if e.session.IsNonceFinished(winner.nonce) || errors.Is(winner.processErr, types.ErrStateHashMismatch) { + return false + } + return winner.err == nil && winner.contentChunks.Load() > 0 && !isFailedStreamAttempt(winner) +} + +type terminalDecision struct { + nonce uint64 + outcome string + stage string + reason string +} + +// resolveTerminalDecision picks the accounting outcome for a request that failed to reach a finished nonce: delivered_pending_settlement when the winner delivered content, otherwise failed. +func (e *Redundancy) resolveTerminalDecision(attempts []*inflight, winnerNonce uint64, failed []*inflight) terminalDecision { + if e.winnerDeliveredPendingSettlement(attempts, winnerNonce) { + return terminalDecision{ + nonce: winnerNonce, + outcome: "delivered_pending_settlement", + stage: "request_delivered_pending_settlement", + reason: "none", + } + } + return terminalDecision{ + nonce: 0, + outcome: "failed", + stage: "request_failed", + reason: gatewayRequestFailureReason(failed), + } +} + func fallbackSuspiciousWinner(attempts []*inflight) *inflight { for _, inf := range attempts { if inf == nil || inf.probe || !inf.suspicious { diff --git a/devshard/cmd/devshardctl/redundancy_delivered_pending_test.go b/devshard/cmd/devshardctl/redundancy_delivered_pending_test.go new file mode 100644 index 0000000000..ac9bb48ad6 --- /dev/null +++ b/devshard/cmd/devshardctl/redundancy_delivered_pending_test.go @@ -0,0 +1,115 @@ +package main + +import ( + "context" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "devshard/types" +) + +// Crowned winner that streamed clean content but never produced an applied finish; old sendTime so the failure branch skips HandleTimeout deterministically. +func deliveredWinnerInflight(nonce uint64) *inflight { + inf := &inflight{ + hostIdx: 0, + hostID: "host-0", + nonce: nonce, + escrowID: "escrow-proxy", + sendTime: time.Now().Add(-longResponseFailureExemption - time.Second), + receiptTime: time.Now().Add(-longResponseFailureExemption), + done: make(chan struct{}), + receiptCh: make(chan struct{}), + firstTokenCh: make(chan struct{}), + } + inf.outputChunks.Store(2) + inf.contentChunks.Store(1) + return inf +} + +func TestWinnerDeliveredPendingSettlement(t *testing.T) { + env := setupTestProxy(t, 2, nil, true) + env.proxy.redundancy.picker.stop() + redundancy := env.proxy.redundancy + + brokeMidStream := deliveredWinnerInflight(7) + brokeMidStream.err = errSimulatedWinnerTransport + + errorStream := deliveredWinnerInflight(7) + errorStream.errorSource = "error.BadRequestError" + + emptyStream := deliveredWinnerInflight(7) + emptyStream.contentChunks.Store(0) + + stateDiverged := deliveredWinnerInflight(7) + stateDiverged.processErr = types.ErrStateHashMismatch + + probeWinner := deliveredWinnerInflight(7) + probeWinner.probe = true + + cases := []struct { + name string + attempts []*inflight + winnerNonce uint64 + want bool + }{ + {name: "delivered_content_not_finished", attempts: []*inflight{deliveredWinnerInflight(7)}, winnerNonce: 7, want: true}, + {name: "no_crowned_winner", attempts: []*inflight{deliveredWinnerInflight(7)}, winnerNonce: 0, want: false}, + {name: "broke_mid_stream", attempts: []*inflight{brokeMidStream}, winnerNonce: 7, want: false}, + {name: "error_stream", attempts: []*inflight{errorStream}, winnerNonce: 7, want: false}, + {name: "empty_stream", attempts: []*inflight{emptyStream}, winnerNonce: 7, want: false}, + {name: "state_hash_mismatch", attempts: []*inflight{stateDiverged}, winnerNonce: 7, want: false}, + {name: "probe_winner", attempts: []*inflight{probeWinner}, winnerNonce: 7, want: false}, + {name: "winner_nonce_absent", attempts: []*inflight{deliveredWinnerInflight(7)}, winnerNonce: 99, want: false}, + } + for _, testCase := range cases { + t.Run(testCase.name, func(t *testing.T) { + require.Equal(t, testCase.want, redundancy.winnerDeliveredPendingSettlement(testCase.attempts, testCase.winnerNonce)) + }) + } +} + +func deliveredPendingSettlementEnv(t *testing.T) func(winner *inflight, opts raceFinishOptions) RequestAccountingRecord { + t.Helper() + env := setupTestProxy(t, 2, nil, true) + env.proxy.redundancy.picker.stop() + + store, err := NewPerfStore(filepath.Join(t.TempDir(), "perf.db")) + require.NoError(t, err) + t.Cleanup(func() { _ = store.Close() }) + + redundancy := env.proxy.redundancy + redundancy.perf = NewPerfTracker(store) + redundancy.devshardID = "escrow-proxy" + + return func(winner *inflight, opts raceFinishOptions) RequestAccountingRecord { + winner.hostID = env.session.HostLabel(0) + ctx, requestID := ensureRequestLogContext(context.Background()) + redundancy.perf.RecordAccountingRequestStart(requestID, "escrow-proxy", "llama", time.Unix(1000, 0)) + require.False(t, env.session.IsNonceFinished(winner.nonce)) + require.Error(t, redundancy.finishRaceOutcome(ctx, []*inflight{winner}, defaultParams(), Decision{Reason: "primary_only"}, winner.nonce, opts)) + record, ok, findErr := redundancy.perf.FindAccountingRequest(requestID, "escrow-proxy") + require.NoError(t, findErr) + require.True(t, ok) + return record + } +} + +// Winner delivered content but no nonce finished, reached via forceTreatAsFailure (the #1387 route): accounted with the real nonce and delivered_pending_settlement, not failed/0. +func TestFinishRaceOutcome_ForcedDeliveredWinnerRecordsPendingSettlement(t *testing.T) { + run := deliveredPendingSettlementEnv(t) + record := run(deliveredWinnerInflight(7), raceFinishOptions{forceTreatAsFailure: true}) + require.Equal(t, "delivered_pending_settlement", record.Outcome) + require.Equal(t, uint64(7), record.WinnerNonce) +} + +func TestFinishRaceOutcome_BrokenWinnerStaysFailed(t *testing.T) { + run := deliveredPendingSettlementEnv(t) + winner := deliveredWinnerInflight(7) + winner.err = errSimulatedWinnerTransport + record := run(winner, raceFinishOptions{forceTreatAsFailure: true}) + require.Equal(t, "failed", record.Outcome) + require.Equal(t, uint64(0), record.WinnerNonce) +} diff --git a/devshard/types/errors.go b/devshard/types/errors.go index 48a926db1a..e4bfcdddc2 100644 --- a/devshard/types/errors.go +++ b/devshard/types/errors.go @@ -37,7 +37,6 @@ var ( ErrSessionNotFinalizing = errors.New("session is not finalizing") ErrInvalidSeedSig = errors.New("seed signature does not recover to slot owner") ErrDuplicateValidation = errors.New("duplicate validation: address already validated this inference") - ErrInvalidStateSig = errors.New("invalid state signature from host") ErrSessionSettlement = errors.New("session is in settlement: seeds are final") ErrInvalidGroup = errors.New("invalid group") ErrEscrowIDMismatch = errors.New("escrow_id does not match session") diff --git a/devshard/user/session.go b/devshard/user/session.go index 9b1a00a69d..da3383d48f 100644 --- a/devshard/user/session.go +++ b/devshard/user/session.go @@ -469,7 +469,7 @@ func (s *Session) processResponse(hostIdx int, resp *host.HostResponse, inferenc } } - // Verify and store state signature. + // Invalid/unexpected-key state sig is a liveness problem, not a dispute (hash already matched): skip it (not in quorum), don't drop the receipt/finish below. if resp.StateSig != nil { expectedAddr := s.group[hostIdx].ValidatorAddress sigContent := &types.StateSignatureContent{ @@ -481,27 +481,26 @@ func (s *Session) processResponse(hostIdx int, resp *host.HostResponse, inferenc if err != nil { return fmt.Errorf("marshal state sig content: %w", err) } - addr, err := s.verifier.RecoverAddress(sigData, resp.StateSig) - if err != nil { - return fmt.Errorf("%w: host %d: %v", types.ErrInvalidStateSig, hostIdx, err) - } - if addr != expectedAddr { - if !s.sm.CheckWarmKey(addr, expectedAddr) { - return fmt.Errorf("%w: host %d: expected %s, got %s", - types.ErrInvalidStateSig, hostIdx, expectedAddr, addr) + addr, recoverErr := s.verifier.RecoverAddress(sigData, resp.StateSig) + switch { + case recoverErr != nil: + logging.Warn("skipping unverifiable state signature (not counted toward quorum)", + "escrow_id", s.escrowID, "host", hostIdx, "nonce", resp.Nonce, "error", recoverErr) + case addr != expectedAddr && !s.sm.CheckWarmKey(addr, expectedAddr): + logging.Warn("skipping state signature from unexpected key (not counted toward quorum)", + "escrow_id", s.escrowID, "host", hostIdx, "nonce", resp.Nonce, "expected", expectedAddr, "got", addr) + default: + // Store for all slots owned by this validator address. + if _, ok := s.signatures[resp.Nonce]; !ok { + s.signatures[resp.Nonce] = make(map[uint32][]byte) } - } - - // Store for all slots owned by this validator address. - if _, ok := s.signatures[resp.Nonce]; !ok { - s.signatures[resp.Nonce] = make(map[uint32][]byte) - } - for _, slot := range s.addrToSlots[expectedAddr] { - s.signatures[resp.Nonce][slot] = resp.StateSig - if s.store != nil { - if sigErr := s.store.AddSignature(s.escrowID, resp.Nonce, slot, resp.StateSig); sigErr != nil { - logging.Warn("failed to persist signature", - "escrow_id", s.escrowID, "nonce", resp.Nonce, "slot", slot, "error", sigErr) + for _, slot := range s.addrToSlots[expectedAddr] { + s.signatures[resp.Nonce][slot] = resp.StateSig + if s.store != nil { + if sigErr := s.store.AddSignature(s.escrowID, resp.Nonce, slot, resp.StateSig); sigErr != nil { + logging.Warn("failed to persist signature", + "escrow_id", s.escrowID, "nonce", resp.Nonce, "slot", slot, "error", sigErr) + } } } } diff --git a/devshard/user/warmkey_test.go b/devshard/user/warmkey_test.go index 724e2dc042..c212398e87 100644 --- a/devshard/user/warmkey_test.go +++ b/devshard/user/warmkey_test.go @@ -134,8 +134,8 @@ func TestProcessResponse_WarmKey_Rejected(t *testing.T) { err = session.ProcessResponse(1, &host.HostResponse{ Nonce: 1, StateHash: root, StateSig: stateSig, }, 1) - require.Error(t, err, "rejected warm key should cause error") - require.ErrorIs(t, err, types.ErrInvalidStateSig) + require.NoError(t, err, "rejected warm key is skipped, not fatal to the response") + require.Empty(t, session.signatures[1], "rejected warm key must not count toward quorum") } func TestProcessResponse_WarmKey_NoResolver(t *testing.T) { @@ -176,8 +176,8 @@ func TestProcessResponse_WarmKey_NoResolver(t *testing.T) { err = session.ProcessResponse(1, &host.HostResponse{ Nonce: 1, StateHash: root, StateSig: stateSig, }, 1) - require.Error(t, err, "without resolver, warm key mismatch should fail") - require.ErrorIs(t, err, types.ErrInvalidStateSig) + require.NoError(t, err, "without resolver, warm key mismatch is skipped, not fatal") + require.Empty(t, session.signatures[1], "unresolved warm key must not count toward quorum") } func TestProcessResponse_ColdKey_StillWorks(t *testing.T) { @@ -188,6 +188,53 @@ func TestProcessResponse_ColdKey_StillWorks(t *testing.T) { require.NotEmpty(t, session.Signatures()) } +// A1: an unverifiable state sig must not drop the executor's receipt/finish — the nonce still finishes and its txs are queued. +func TestProcessResponse_InvalidStateSig_StillProcessesReceiptAndFinish(t *testing.T) { + coldKeys := makeKeys(t, 3) + warmKeys := makeKeys(t, 3) + userKey := testutil.MustGenerateKey(t) + group := testutil.MakeGroup(coldKeys) + config := testutil.DefaultConfig(3) + verifier := signing.NewSecp256k1Verifier() + + userSM, err := state.NewStateMachine("escrow-1", config, group, 100000, userKey.Address(), verifier, testutil.MustMemoryStore(t, "escrow-1", userKey.Address(), config, group, 100000)) + require.NoError(t, err) + + root, err := userSM.ApplyLocal(1, []*types.DevshardTx{testutil.StartTx(1)}) + require.NoError(t, err) + + sigData, err := proto.Marshal(&types.StateSignatureContent{StateRoot: root, EscrowId: "escrow-1", Nonce: 1}) + require.NoError(t, err) + badSig, err := warmKeys[1].Sign(sigData) // recovers to a warm key with no resolver -> rejected + require.NoError(t, err) + + clients := make([]HostClient, 3) + for i := range clients { + clients[i] = &ErrorClient{} + } + session, err := NewSession(userSM, userKey, "escrow-1", group, clients, verifier) + require.NoError(t, err) + session.nonce = 1 + session.diffs = append(session.diffs, types.Diff{Nonce: 1, PostStateRoot: root}) + session.nonceStates[1] = &nonceOutcome{} + + finishTx := &types.DevshardTx{Tx: &types.DevshardTx_FinishInference{ + FinishInference: &types.MsgFinishInference{InferenceId: 1}, + }} + err = session.ProcessResponse(1, &host.HostResponse{ + Nonce: 1, + StateHash: root, + StateSig: badSig, + Receipt: []byte("receipt"), + Mempool: []*types.DevshardTx{finishTx}, + }, 1) + + require.NoError(t, err, "invalid state sig must not drop the response") + require.Empty(t, session.signatures[1], "invalid sig must not count toward quorum") + require.True(t, session.IsNonceFinished(1), "finish must still be observed after a skipped sig") + require.NotEmpty(t, session.PendingTxs(), "receipt and finish must still be queued") +} + func TestProcessResponse_WarmKey_Finalize(t *testing.T) { session, _, _ := setupWarmKeySession(t, 3)