fix(blockvalidation): an in-flight tagging run no longer overwrites an invalidation's mined_set reset

services/blockassembly/blockassembly_system_test.go

@@ -1181,7 +1181,7 @@ func TestHandleReorgWithInvalidBlock_Integration(t *testing.T) {
 	// so call SetBlockMinedSet directly — otherwise BA's reset() blocks in
 	// waitForBlockMinedSet until its retry budget exhausts.
 	for i := range invalidatedHashes {
-		require.NoError(t, ba.blockchainClient.SetBlockMinedSet(ctx, &invalidatedHashes[i]),
+		require.NoError(t, ba.blockchainClient.SetBlockMinedSet(ctx, &invalidatedHashes[i], true),
 			"failed to re-set mined_set on invalidated block %s", invalidatedHashes[i].String())
 	}
 

services/blockchain/Client.go

@@ -1683,12 +1683,15 @@ func (c *Client) GetBlockIsMined(ctx context.Context, blockHash *chainhash.Hash)
 // Parameters:
 //   - ctx: Context for the operation with timeout and cancellation support
 //   - blockHash: Hash of the block to mark as mined
+//   - expectInvalid: The block's invalid flag when the caller's tagging run started
 //
 // Returns:
-//   - error: Any error encountered during the mining status update operation
-func (c *Client) SetBlockMinedSet(ctx context.Context, blockHash *chainhash.Hash) error {
+//   - error: errors.ErrStateError if the block's invalid flag has changed, or any
+//     other error encountered during the mining status update operation
+func (c *Client) SetBlockMinedSet(ctx context.Context, blockHash *chainhash.Hash, expectInvalid bool) error {
 	_, err := c.client.SetBlockMinedSet(ctx, &blockchain_api.SetBlockMinedSetRequest{
-		BlockHash: blockHash[:],
+		BlockHash:     blockHash[:],
+		ExpectInvalid: expectInvalid,
 	})
 	if err != nil {
 		return errors.UnwrapGRPC(err)

services/blockchain/Interface.go

@@ -602,10 +602,14 @@ type ClientI interface {
 	// Parameters:
 	// - ctx: Context for the operation with timeout and cancellation support
 	// - blockHash: Hash of the block to mark as mined
+	// - expectInvalid: The block's invalid flag when the caller's tagging run started.
+	//   The write is refused if the flag has changed since, because InvalidateBlock or
+	//   RevalidateBlock cleared mined_set so the block is processed again.
 	//
 	// Returns:
+	// - errors.ErrStateError if the block's invalid flag no longer equals expectInvalid
 	// - Error if the operation fails, nil on success
-	SetBlockMinedSet(ctx context.Context, blockHash *chainhash.Hash) error
+	SetBlockMinedSet(ctx context.Context, blockHash *chainhash.Hash, expectInvalid bool) error
 
 	// ClearBlockMinedSet resets the mined_set flag to false for a block.
 	//

services/blockchain/LocalClient.go

@@ -388,8 +388,8 @@ func (c *LocalClient) GetBlockIsMined(ctx context.Context, blockHash *chainhash.
 	return c.store.GetBlockIsMined(ctx, blockHash)
 }
 
-func (c *LocalClient) SetBlockMinedSet(ctx context.Context, blockHash *chainhash.Hash) error {
-	return c.store.SetBlockMinedSet(ctx, blockHash)
+func (c *LocalClient) SetBlockMinedSet(ctx context.Context, blockHash *chainhash.Hash, expectInvalid bool) error {
+	return c.store.SetBlockMinedSet(ctx, blockHash, expectInvalid)
 }
 
 func (c *LocalClient) ClearBlockMinedSet(ctx context.Context, blockHash *chainhash.Hash) error {

services/blockchain/LocalClient_test.go

@@ -370,7 +370,7 @@ func TestLocalClient_NilStorePanics(t *testing.T) {
 		{
 			name: "SetBlockMinedSet",
 			fn: func() {
-				_ = client.SetBlockMinedSet(ctx, &blockHash)
+				_ = client.SetBlockMinedSet(ctx, &blockHash, false)
 			},
 		},
 		{

services/blockchain/Server.go

@@ -2978,7 +2978,7 @@ func (b *Blockchain) SetBlockMinedSet(ctx context.Context, req *blockchain_api.S
 		return nil, errors.WrapGRPC(errors.NewInvalidArgumentError("[Blockchain][SetBlockMinedSet] hash must be %d bytes, got %d", chainhash.HashSize, len(req.BlockHash)))
 	}
 
-	err = b.store.SetBlockMinedSet(ctx, blockHash)
+	err = b.store.SetBlockMinedSet(ctx, blockHash, req.ExpectInvalid)
 	if err != nil {
 		return nil, errors.WrapGRPC(err)
 	}

services/blockchain/blockchain_api/blockchain_api.proto

@@ -618,6 +618,9 @@ message GetNextWorkRequiredResponse {
 // SetBlockMinedSetRequest marks a block as mined.
 message SetBlockMinedSetRequest {
   bytes blockHash = 1;  // Hash of the mined block
+  // The block's invalid flag when the caller's tagging run started; the write is
+  // refused with ERR_STATE_ERROR if the flag has changed since.
+  bool expectInvalid = 2;
 }
 
 message ClearBlockMinedSetRequest {

services/blockchain/client_test.go

@@ -4365,7 +4365,7 @@ func TestClientSetBlockMinedSet(t *testing.T) {
 			settings: tSettings,
 		}
 
-		err := c.SetBlockMinedSet(ctx, &blockHash)
+		err := c.SetBlockMinedSet(ctx, &blockHash, false)
 		require.NoError(t, err)
 	})
 
@@ -4379,10 +4379,27 @@ func TestClientSetBlockMinedSet(t *testing.T) {
 			settings: tSettings,
 		}
 
-		err := c.SetBlockMinedSet(ctx, &blockHash)
+		err := c.SetBlockMinedSet(ctx, &blockHash, false)
 		require.Error(t, err)
 		assert.Contains(t, err.Error(), "grpc error")
 	})
+
+	t.Run("refused write keeps ErrStateError across gRPC", func(t *testing.T) {
+		// setTxMinedStatus re-queues the block only if errors.Is(err, ErrStateError)
+		// still holds after the server's WrapGRPC and the client's UnwrapGRPC
+		c := &Client{
+			client: &mockBlockClient{
+				err: errors.WrapGRPC(errors.NewStateError("invalid flag changed")),
+			},
+			logger:   logger,
+			settings: tSettings,
+		}
+
+		err := c.SetBlockMinedSet(ctx, &blockHash, true)
+		require.Error(t, err)
+		require.True(t, errors.Is(err, errors.ErrStateError), "expected ErrStateError, got %v", err)
+		require.True(t, c.client.(*mockBlockClient).lastSetBlockMinedSetReq.ExpectInvalid, "expectInvalid must reach the request")
+	})
 }
 
 // TestClientSetBlockProcessedAt tests the SetBlockProcessedAt method

services/blockchain/mock.go

@@ -379,8 +379,8 @@ func (m *Mock) SetState(ctx context.Context, key string, data []byte) error {
 }
 
 // SetBlockMinedSet mocks the SetBlockMinedSet method
-func (m *Mock) SetBlockMinedSet(ctx context.Context, blockHash *chainhash.Hash) error {
-	args := m.Called(ctx, blockHash)
+func (m *Mock) SetBlockMinedSet(ctx context.Context, blockHash *chainhash.Hash, expectInvalid bool) error {
+	args := m.Called(ctx, blockHash, expectInvalid)
 	return args.Error(0)
 }
 
@@ -770,8 +770,8 @@ func (m *mockStoreGetSetBlockIsMined) GetBlockIsMined(ctx context.Context, hash
 	return args.Bool(0), nil
 }
 
-func (m *mockStoreGetSetBlockIsMined) SetBlockMinedSet(ctx context.Context, hash *chainhash.Hash) error {
-	args := m.Called(ctx, hash)
+func (m *mockStoreGetSetBlockIsMined) SetBlockMinedSet(ctx context.Context, hash *chainhash.Hash, expectInvalid bool) error {
+	args := m.Called(ctx, hash, expectInvalid)
 	return args.Error(0)
 }
 

services/blockchain/server_test.go

@@ -2543,7 +2543,7 @@ func TestSetBlockMinedSet(t *testing.T) {
 	t.Run("error - store returns error", func(t *testing.T) {
 		hash := chainhash.DoubleHashH([]byte("error-case"))
 		mockStore := &mockStoreGetSetBlockIsMined{}
-		mockStore.On("SetBlockMinedSet", mock.Anything, &hash).
+		mockStore.On("SetBlockMinedSet", mock.Anything, &hash, false).
 			Return(errors.NewProcessingError("forced error"))
 
 		server, _ := New(ctx, logger, tSettings, mockStore, nil)
@@ -2561,7 +2561,7 @@ func TestSetBlockMinedSet(t *testing.T) {
 	t.Run("success - block marked as mined", func(t *testing.T) {
 		hash := chainhash.DoubleHashH([]byte("success-case"))
 		mockStore := &mockStoreGetSetBlockIsMined{}
-		mockStore.On("SetBlockMinedSet", mock.Anything, &hash).
+		mockStore.On("SetBlockMinedSet", mock.Anything, &hash, false).
 			Return(nil)
 
 		server, _ := New(ctx, logger, tSettings, mockStore, nil)

services/blockpersister/Server_test.go

@@ -759,7 +759,7 @@ func (m *MockBlockchainClient) GetMedianTimePastRange(_ context.Context, fromHei
 func (m *MockBlockchainClient) SetState(ctx context.Context, key string, data []byte) error {
 	return nil
 }
-func (m *MockBlockchainClient) SetBlockMinedSet(ctx context.Context, blockHash *chainhash.Hash) error {
+func (m *MockBlockchainClient) SetBlockMinedSet(ctx context.Context, blockHash *chainhash.Hash, expectInvalid bool) error {
 	return nil
 }
 func (m *MockBlockchainClient) ClearBlockMinedSet(ctx context.Context, blockHash *chainhash.Hash) error {

services/blockvalidation/BlockValidation.go

@@ -1568,8 +1568,22 @@ func (u *BlockValidation) setTxMinedStatus(ctx context.Context, blockHash *chain
 		u.logger.Warnf("[setTxMined][%s] failed closing subtrees after setting mined: %v", block.Hash().String(), releaseErr)
 	}
 
-	// update block mined_set to true
-	if err = u.blockchainClient.SetBlockMinedSet(ctx, blockHash); err != nil {
+	// update block mined_set to true, but only if the block's invalid flag is still the one
+	// this run started from. InvalidateBlock and RevalidateBlock flip that flag and clear
+	// mined_set so the block is processed again; overwriting mined_set here would make the
+	// worker's MinedSet guard and the mined-not-set sweep both skip the block for good,
+	// leaving an invalid block tagged or a revalidated block untagged.
+	startedInvalid := len(unsetMined) > 0 && unsetMined[0]
+	if err = u.blockchainClient.SetBlockMinedSet(ctx, blockHash, startedInvalid); err != nil {
+		if errors.Is(err, errors.ErrStateError) {
+			// Not a failure of this run: re-queue so the block is processed again under its
+			// new invalid flag, and return nil so the retry counter is not charged.
+			u.logger.Warnf("[setTxMined][%s] block invalid flag changed during the run, re-queueing: %v", blockHash.String(), err)
+			u.enqueueSetMined(blockHash)
+
+			return nil
+		}
+
 		return errors.NewServiceError("[setTxMined][%s] failed to set block mined", block.Hash().String(), err)
 	}
 

services/blockvalidation/BlockValidation_test.go

@@ -241,13 +241,13 @@ func (t *trackingBlockchainClient) Subscribe(ctx context.Context, tag string) (c
 	return t.ClientI.Subscribe(ctx, tag)
 }
 
-func (t *trackingBlockchainClient) SetBlockMinedSet(ctx context.Context, hash *chainhash.Hash) error {
+func (t *trackingBlockchainClient) SetBlockMinedSet(ctx context.Context, hash *chainhash.Hash, expectInvalid bool) error {
 	if t.setMinedCalled != nil {
 		t.setMinedOnce.Do(func() {
 			close(t.setMinedCalled)
 		})
 	}
-	return t.ClientI.SetBlockMinedSet(ctx, hash)
+	return t.ClientI.SetBlockMinedSet(ctx, hash, expectInvalid)
 }
 
 func (m *MockSubtreeValidationClient) Health(ctx context.Context, checkLiveness bool) (int, string, error) {

services/blockvalidation/InvalidBlocksKafkaProducer_test.go

@@ -400,8 +400,8 @@ func (m *MockBlockchainClient) SendNotification(ctx context.Context, notificatio
 }
 
 // SetBlockMinedSet implements the blockchain.ClientI interface
-func (m *MockBlockchainClient) SetBlockMinedSet(ctx context.Context, blockHash *chainhash.Hash) error {
-	args := m.Called(ctx, blockHash)
+func (m *MockBlockchainClient) SetBlockMinedSet(ctx context.Context, blockHash *chainhash.Hash, expectInvalid bool) error {
+	args := m.Called(ctx, blockHash, expectInvalid)
 	return args.Error(0)
 }
 

services/blockvalidation/cached_headers_median_window_test.go

@@ -323,7 +323,7 @@ func runHeaderContextRetryReachesAddBlock(t *testing.T, disableOptimisticMining
 	mockBlockchain.On("SetBlockSubtreesSet", mock.Anything, mock.Anything).Return(nil).Maybe()
 	mockBlockchain.On("GetBlockHeaderIDs", mock.Anything, mock.Anything, mock.Anything).
 		Return([]uint32{1, 0}, nil).Maybe()
-	mockBlockchain.On("SetBlockMinedSet", mock.Anything, mock.Anything).Return(nil).Maybe()
+	mockBlockchain.On("SetBlockMinedSet", mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe()
 	mockBlockchain.On("InvalidateBlock", mock.Anything, mock.Anything).Return([]chainhash.Hash{}, nil).Maybe()
 	mockBlockchain.On("GetBlockHeadersFromHeight", mock.Anything, mock.Anything, mock.Anything).
 		Return(goodRun, goodMeta, nil).Maybe()

services/blockvalidation/checkpoint_difficulty_reconsider_test.go

@@ -113,7 +113,7 @@ func TestReconsider_BelowCheckpoint_RunsExpectedNBitsAndSucceeds(t *testing.T) {
 	// Reconsider the ancestor. RevalidateBlock clears mined_set, so re-stamp it the way the
 	// setMined worker would; the descendant's validation waits on the parent's mined state.
 	require.NoError(t, client.RevalidateBlock(ctx, parent.Hash()))
-	require.NoError(t, client.SetBlockMinedSet(ctx, parent.Hash()))
+	require.NoError(t, client.SetBlockMinedSet(ctx, parent.Hash(), false))
 
 	// The difficulty inputs must still be reachable after the invalidation, which is the whole
 	// claim: the calculator walks parent_id and does not filter invalid rows.

services/blockvalidation/corrupt_escalation_test.go

@@ -261,7 +261,7 @@ func TestOptimisticCorrupt_ReValidationConvergesOnInvalidation(t *testing.T) {
 
 	// waitForPreviousBlocksToBeProcessed gates on the parent being mined, and the store writes
 	// genesis unmined; without this the optimistic AddBlock is never reached.
-	require.NoError(t, realClient.SetBlockMinedSet(ctx, tSettings.ChainCfgParams.GenesisHash))
+	require.NoError(t, realClient.SetBlockMinedSet(ctx, tSettings.ChainCfgParams.GenesisHash, false))
 
 	blockchainClient := &invalidateFailsFirstNClient{ClientI: realClient, failFirst: 1}
 
@@ -867,7 +867,7 @@ func TestOptimisticCorrupt_StrikeGatedOnRevalidation(t *testing.T) {
 
 		// waitForPreviousBlocksToBeProcessed gates on the parent being mined, and the store writes
 		// genesis unmined; without this the optimistic AddBlock is never reached.
-		require.NoError(t, realClient.SetBlockMinedSet(ctx, tSettings.ChainCfgParams.GenesisHash))
+		require.NoError(t, realClient.SetBlockMinedSet(ctx, tSettings.ChainCfgParams.GenesisHash, false))
 
 		// Height 1, so the model height, the store-derived height and the BIP34 coinbase-encoded
 		// height all agree (StoreBlock re-derives the height from the parent chain).