vault: add per-call 2s timeout to BroadcastBlob invocations

Each parallel BroadcastBlob call now gets a 2-second timeout derived from
the parent context. A slow individual broadcast will be cancelled and
skipped without stalling the rest of the batch. Parent context
cancellation still propagates immediately for round-level failures.

Made-with: Cursor

Author: prashantkumar1982

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

@@ -543,10 +543,12 @@ func (r *ReportingPlugin) Observation(ctx context.Context, seqNr uint64, aq type
 
 // broadcastBlobPayloads broadcasts each payload as a blob in parallel to reduce
 // 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. Context cancellation/deadline errors are
-// propagated immediately so that expired rounds fail fast.
+// DeltaProgress). Each call is given a 2-second timeout so that a single slow
+// broadcast cannot stall the entire batch. 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. Context
+// cancellation/deadline errors on the parent context are propagated immediately so
+// that expired rounds fail fast.
 func (r *ReportingPlugin) broadcastBlobPayloads(
 	ctx context.Context,
 	fetcher ocr3_1types.BlobBroadcastFetcher,
@@ -561,10 +563,14 @@ func (r *ReportingPlugin) broadcastBlobPayloads(
 		r.lggr.Debugw("observation blob broadcast finished", "seqNr", seqNr, "blobCount", len(payloads), "elapsed", time.Since(start))
 	}()
 
+	const perBlobTimeout = 2 * time.Second
 	var g errgroup.Group
 	for i, payload := range payloads {
 		g.Go(func() error {
-			blobHandle, err := fetcher.BroadcastBlob(ctx, payload, ocr3_1types.BlobExpirationHintSequenceNumber{SeqNr: seqNr + 2})
+			broadcastCtx, cancel := context.WithTimeout(ctx, perBlobTimeout)
+			defer cancel()
+
+			blobHandle, err := fetcher.BroadcastBlob(broadcastCtx, payload, ocr3_1types.BlobExpirationHintSequenceNumber{SeqNr: seqNr + 2})
 			if err != nil {
 				if ctx.Err() != nil {
 					return ctx.Err()

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

@@ -5188,6 +5188,21 @@ func (f *callbackBlobFetcher) FetchBlob(context.Context, ocr3_1types.BlobHandle)
 	panic("FetchBlob should not be called in broadcastBlobPayloads tests")
 }
 
+type ctxCallbackBlobFetcher struct {
+	fn func(ctx context.Context, payload []byte) error
+}
+
+func (f *ctxCallbackBlobFetcher) BroadcastBlob(ctx context.Context, payload []byte, _ ocr3_1types.BlobExpirationHint) (ocr3_1types.BlobHandle, error) {
+	if err := f.fn(ctx, payload); err != nil {
+		return ocr3_1types.BlobHandle{}, err
+	}
+	return ocr3_1types.BlobHandle{}, nil
+}
+
+func (f *ctxCallbackBlobFetcher) FetchBlob(context.Context, ocr3_1types.BlobHandle) ([]byte, error) {
+	panic("FetchBlob should not be called in broadcastBlobPayloads tests")
+}
+
 func TestPlugin_StateTransition_StoresPendingQueue(t *testing.T) {
 	lggr := logger.TestLogger(t)
 	store := requests.NewStore[*vaulttypes.Request]()
@@ -7332,4 +7347,35 @@ func TestPlugin_broadcastBlobPayloads(t *testing.T) {
 		assert.Nil(t, result)
 		assert.ErrorIs(t, err, context.DeadlineExceeded)
 	})
+
+	t.Run("slow broadcast hits per-call timeout and is skipped", func(t *testing.T) {
+		lggr, observed := logger.TestLoggerObserved(t, zapcore.WarnLevel)
+		r := &ReportingPlugin{
+			lggr:    lggr,
+			metrics: newTestMetrics(t),
+			marshalBlob: func(ocr3_1types.BlobHandle) ([]byte, error) {
+				return []byte("handle"), nil
+			},
+		}
+
+		fetcher := &ctxCallbackBlobFetcher{fn: func(ctx context.Context, payload []byte) error {
+			if string(payload) == "slow" {
+				<-ctx.Done()
+				return ctx.Err()
+			}
+			return nil
+		}}
+
+		payloads := [][]byte{[]byte("fast"), []byte("slow")}
+		ids := []string{"req-fast", "req-slow"}
+
+		result, err := r.broadcastBlobPayloads(t.Context(), fetcher, 1, payloads, ids)
+		require.NoError(t, err)
+		assert.Len(t, result, 1)
+
+		warnLogs := observed.FilterMessage("failed to broadcast pending queue item as blob, skipping")
+		assert.Equal(t, 1, warnLogs.Len())
+		fields := warnLogs.All()[0].ContextMap()
+		assert.Equal(t, "req-slow", fields["requestID"])
+	})
 }