vault: propagate context cancellation from blob broadcast failures

Check ctx.Err() when BroadcastBlob fails so that context.Canceled and
context.DeadlineExceeded are returned immediately rather than swallowed.
This preserves fail-fast semantics for expired OCR rounds while still
skipping item-specific transient errors.

Made-with: Cursor

Author: prashantkumar1982

core/services/ocr2/plugins/vault/plugin.go

@@ -514,7 +514,11 @@ func (r *ReportingPlugin) Observation(ctx context.Context, seqNr uint64, aq type
 		}
 	}
 
-	obspb.PendingQueueItems = r.broadcastBlobPayloads(ctx, blobBroadcastFetcher, seqNr, blobPayloads, blobPayloadIDs)
+	pendingQueueItems, err := r.broadcastBlobPayloads(ctx, blobBroadcastFetcher, seqNr, blobPayloads, blobPayloadIDs)
+	if err != nil {
+		return nil, err
+	}
+	obspb.PendingQueueItems = pendingQueueItems
 
 	// Second, generate a random nonce that we'll use to sort the observations.
 	// Each node generates a nonce idepedently, to be concatenated later on.
@@ -541,14 +545,15 @@ func (r *ReportingPlugin) Observation(ctx context.Context, seqNr uint64, aq type
 // Observation() latency (shortening this phase helps the OCR round finish within
 // DeltaProgress). Individual broadcast failures are logged and skipped rather than
 // aborting the entire observation, so that one problematic payload does not prevent
-// the remaining items from being observed.
+// the remaining items from being observed. Context cancellation/deadline errors are
+// propagated immediately so that expired rounds fail fast.
 func (r *ReportingPlugin) broadcastBlobPayloads(
 	ctx context.Context,
 	fetcher ocr3_1types.BlobBroadcastFetcher,
 	seqNr uint64,
 	payloads [][]byte,
 	requestIDs []string,
-) [][]byte {
+) ([][]byte, error) {
 	results := make([][]byte, len(payloads))
 
 	start := time.Now()
@@ -561,6 +566,9 @@ func (r *ReportingPlugin) broadcastBlobPayloads(
 		g.Go(func() error {
 			blobHandle, err := fetcher.BroadcastBlob(ctx, payload, ocr3_1types.BlobExpirationHintSequenceNumber{SeqNr: seqNr + 2})
 			if err != nil {
+				if ctx.Err() != nil {
+					return ctx.Err()
+				}
 				r.lggr.Warnw("failed to broadcast pending queue item as blob, skipping",
 					"seqNr", seqNr,
 					"requestID", requestIDs[i],
@@ -581,15 +589,17 @@ func (r *ReportingPlugin) broadcastBlobPayloads(
 			return nil
 		})
 	}
-	_ = g.Wait()
+	if err := g.Wait(); err != nil {
+		return nil, err
+	}
 
 	filtered := make([][]byte, 0, len(results))
 	for _, item := range results {
 		if item != nil {
 			filtered = append(filtered, item)
 		}
 	}
-	return filtered
+	return filtered, nil
 }
 
 func (r *ReportingPlugin) observeGetSecrets(ctx context.Context, reader ReadKVStore, req proto.Message, o *vaultcommon.Observation) {

core/services/ocr2/plugins/vault/plugin_test.go

@@ -7143,7 +7143,8 @@ func TestPlugin_broadcastBlobPayloads(t *testing.T) {
 		}
 
 		fetcher := &callbackBlobFetcher{fn: func([]byte) error { return nil }}
-		result := r.broadcastBlobPayloads(t.Context(), fetcher, 1, nil, nil)
+		result, err := r.broadcastBlobPayloads(t.Context(), fetcher, 1, nil, nil)
+		require.NoError(t, err)
 		assert.Empty(t, result)
 	})
 
@@ -7161,7 +7162,8 @@ func TestPlugin_broadcastBlobPayloads(t *testing.T) {
 		payloads := [][]byte{[]byte("p1"), []byte("p2"), []byte("p3")}
 		ids := []string{"req-1", "req-2", "req-3"}
 
-		result := r.broadcastBlobPayloads(t.Context(), fetcher, 1, payloads, ids)
+		result, err := r.broadcastBlobPayloads(t.Context(), fetcher, 1, payloads, ids)
+		require.NoError(t, err)
 		assert.Len(t, result, 3)
 		for _, item := range result {
 			assert.Equal(t, []byte("handle"), item)
@@ -7188,7 +7190,8 @@ func TestPlugin_broadcastBlobPayloads(t *testing.T) {
 		payloads := [][]byte{[]byte("p1"), []byte("p2"), []byte("p3")}
 		ids := []string{"req-1", "req-2", "req-3"}
 
-		result := r.broadcastBlobPayloads(t.Context(), fetcher, 5, payloads, ids)
+		result, err := r.broadcastBlobPayloads(t.Context(), fetcher, 5, payloads, ids)
+		require.NoError(t, err)
 		assert.Len(t, result, 2)
 
 		warnLogs := observed.FilterMessage("failed to broadcast pending queue item as blob, skipping")
@@ -7213,7 +7216,8 @@ func TestPlugin_broadcastBlobPayloads(t *testing.T) {
 		payloads := [][]byte{[]byte("p1"), []byte("p2")}
 		ids := []string{"req-1", "req-2"}
 
-		result := r.broadcastBlobPayloads(t.Context(), fetcher, 1, payloads, ids)
+		result, err := r.broadcastBlobPayloads(t.Context(), fetcher, 1, payloads, ids)
+		require.NoError(t, err)
 		assert.Empty(t, result)
 
 		warnLogs := observed.FilterMessage("failed to broadcast pending queue item as blob, skipping")
@@ -7234,7 +7238,8 @@ func TestPlugin_broadcastBlobPayloads(t *testing.T) {
 		payloads := [][]byte{[]byte("p1"), []byte("p2")}
 		ids := []string{"req-1", "req-2"}
 
-		result := r.broadcastBlobPayloads(t.Context(), fetcher, 1, payloads, ids)
+		result, err := r.broadcastBlobPayloads(t.Context(), fetcher, 1, payloads, ids)
+		require.NoError(t, err)
 		assert.Empty(t, result)
 
 		warnLogs := observed.FilterMessage("failed to marshal blob handle, skipping")
@@ -7267,12 +7272,64 @@ func TestPlugin_broadcastBlobPayloads(t *testing.T) {
 		payloads := [][]byte{[]byte("p1"), []byte("p2"), []byte("p3")}
 		ids := []string{"req-1", "req-2", "req-3"}
 
-		result := r.broadcastBlobPayloads(t.Context(), fetcher, 1, payloads, ids)
+		result, err := r.broadcastBlobPayloads(t.Context(), fetcher, 1, payloads, ids)
+		require.NoError(t, err)
 
 		broadcastWarns := observed.FilterMessage("failed to broadcast pending queue item as blob, skipping")
 		marshalWarns := observed.FilterMessage("failed to marshal blob handle, skipping")
 		assert.Equal(t, 1, broadcastWarns.Len())
 		assert.Equal(t, 1, marshalWarns.Len())
 		assert.Len(t, result, 1)
 	})
+
+	t.Run("context cancellation propagates error", func(t *testing.T) {
+		lggr := logger.TestLogger(t)
+		r := &ReportingPlugin{
+			lggr:    lggr,
+			metrics: newTestMetrics(t),
+			marshalBlob: func(ocr3_1types.BlobHandle) ([]byte, error) {
+				return []byte("handle"), nil
+			},
+		}
+
+		ctx, cancel := context.WithCancel(t.Context())
+		cancel()
+
+		fetcher := &callbackBlobFetcher{fn: func([]byte) error {
+			return ctx.Err()
+		}}
+
+		payloads := [][]byte{[]byte("p1"), []byte("p2")}
+		ids := []string{"req-1", "req-2"}
+
+		result, err := r.broadcastBlobPayloads(ctx, fetcher, 1, payloads, ids)
+		assert.Nil(t, result)
+		assert.ErrorIs(t, err, context.Canceled)
+	})
+
+	t.Run("context deadline exceeded propagates error", func(t *testing.T) {
+		lggr := logger.TestLogger(t)
+		r := &ReportingPlugin{
+			lggr:    lggr,
+			metrics: newTestMetrics(t),
+			marshalBlob: func(ocr3_1types.BlobHandle) ([]byte, error) {
+				return []byte("handle"), nil
+			},
+		}
+
+		ctx, cancel := context.WithTimeout(t.Context(), 0)
+		defer cancel()
+		<-ctx.Done()
+
+		fetcher := &callbackBlobFetcher{fn: func([]byte) error {
+			return ctx.Err()
+		}}
+
+		payloads := [][]byte{[]byte("p1")}
+		ids := []string{"req-1"}
+
+		result, err := r.broadcastBlobPayloads(ctx, fetcher, 1, payloads, ids)
+		assert.Nil(t, result)
+		assert.ErrorIs(t, err, context.DeadlineExceeded)
+	})
 }