Alpha support for out-of-consensus key-value state reads

networking/bootstrapper_v2.go

@@ -61,9 +61,11 @@ func newBootstrapperV2(
 }
 
 func (b *bootstrapperV2) Start() error {
-	succeeded := false
+	// Armed only once we've transitioned to started, so that a rejected Start()
+	// never tears down a bootstrapperV2 that someone else started.
+	needsTeardown := false
 	defer func() {
-		if !succeeded {
+		if needsTeardown {
 			b.Close()
 		}
 	}()
@@ -76,9 +78,10 @@ func (b *bootstrapperV2) Start() error {
 	}
 
 	b.state = bootstrapperStarted
+	needsTeardown = true
 
 	b.logger.Info("BootstrapperV2: Started listening", nil)
-	succeeded = true
+	needsTeardown = false
 	return nil
 }
 

networking/ocr_endpoint_v2.go

@@ -148,9 +148,11 @@ func streamNameFromConfigDigest(cd ocr2types.ConfigDigest) string {
 
 // Start the ocrEndpointV2. Should only be called once.
 func (o *ocrEndpointV2) Start() error {
-	succeeded := false
+	// Armed only once we've transitioned to started, so that a rejected Start()
+	// never tears down an ocrEndpointV2 that someone else started.
+	needsTeardown := false
 	defer func() {
-		if !succeeded {
+		if needsTeardown {
 			o.Close()
 		}
 	}()
@@ -162,6 +164,7 @@ func (o *ocrEndpointV2) Start() error {
 		return fmt.Errorf("cannot start ocrEndpointV2 that is not unstarted, state was: %d", o.state)
 	}
 	o.state = ocrEndpointStarted
+	needsTeardown = true
 
 	for oid, pid := range o.peerMapping {
 		if oid == o.ownOracleID {
@@ -199,7 +202,7 @@ func (o *ocrEndpointV2) Start() error {
 	})
 
 	o.logger.Info("OCREndpointV2: Started listening", nil)
-	succeeded = true
+	needsTeardown = false
 	return nil
 }
 

networking/ocr_endpoint_v3.go

@@ -132,9 +132,11 @@ func newOCREndpointV3(
 
 // Start the ocrEndpointV3. Called once at the end of the initialization code.
 func (o *ocrEndpointV3) start() error {
-	succeeded := false
+	// Armed only once we've transitioned to started, so that a rejected start()
+	// never tears down an ocrEndpointV3 that someone else started.
+	needsTeardown := false
 	defer func() {
-		if !succeeded {
+		if needsTeardown {
 			o.Close()
 		}
 	}()
@@ -146,6 +148,7 @@ func (o *ocrEndpointV3) start() error {
 		return fmt.Errorf("cannot start ocrEndpointV3 that is not unstarted, state was: %d", o.state)
 	}
 	o.state = ocrEndpointStarted
+	needsTeardown = true
 
 	for oid, pid := range o.peerMapping {
 		if oid == o.ownOracleID {
@@ -209,7 +212,7 @@ func (o *ocrEndpointV3) start() error {
 	})
 
 	o.logger.Info("OCREndpointV3: Started listening", nil)
-	succeeded = true
+	needsTeardown = false
 	return nil
 }
 

networking/ragedisco/discovery_protocol.go

@@ -129,9 +129,11 @@ func newDiscoveryProtocol(
 }
 
 func (p *discoveryProtocol) Start() error {
-	succeeded := false
+	// Armed only once we've transitioned to started, so that a rejected Start()
+	// never tears down a discoveryProtocol that someone else started.
+	needsTeardown := false
 	defer func() {
-		if !succeeded {
+		if needsTeardown {
 			p.Close()
 		}
 	}()
@@ -142,6 +144,7 @@ func (p *discoveryProtocol) Start() error {
 		return fmt.Errorf("cannot start discoveryProtocol that is not unstarted, state was: %v", p.state)
 	}
 	p.state = discoveryProtocolStarted
+	needsTeardown = true
 
 	p.lock.Lock()
 	defer p.lock.Unlock()
@@ -153,7 +156,7 @@ func (p *discoveryProtocol) Start() error {
 	p.processes.Go(p.sendLoop)
 	p.processes.Go(p.saveLoop)
 	p.processes.Go(p.statusReportLoop)
-	succeeded = true
+	needsTeardown = false
 	return nil
 }
 

networking/ragedisco/ragep2p_discoverer.go

@@ -79,9 +79,11 @@ func NewRagep2pDiscoverer(
 }
 
 func (r *Ragep2pDiscoverer) Start(host ragep2pwrapper.Host, keyring ragetypes.PeerKeyring, logger loghelper.LoggerWithContext) error {
-	succeeded := false
+	// Armed only once we've transitioned to started, so that a rejected Start()
+	// never tears down a Ragep2pDiscoverer that someone else started.
+	needsTeardown := false
 	defer func() {
-		if !succeeded {
+		if needsTeardown {
 			r.Close()
 		}
 	}()
@@ -93,6 +95,7 @@ func (r *Ragep2pDiscoverer) Start(host ragep2pwrapper.Host, keyring ragetypes.Pe
 		return fmt.Errorf("cannot start Ragep2pDiscoverer that is not unstarted, state was: %v", r.state)
 	}
 	r.state = ragep2pDiscovererStarted
+	needsTeardown = true
 	r.host = host
 	announceAddresses, ok := combinedAnnounceAddrsForDiscoverer(r.logger, r.announceAddresses)
 	if !ok {
@@ -120,7 +123,7 @@ func (r *Ragep2pDiscoverer) Start(host ragep2pwrapper.Host, keyring ragetypes.Pe
 	r.proc.Go(r.connectivityLoop)
 	r.proc.Go(r.writeLoop)
 
-	succeeded = true
+	needsTeardown = false
 	return nil
 }
 

offchainreporting2/reportingplugin/median/median.go

@@ -347,34 +347,34 @@ func (fac NumericalMedianFactory) NewReportingPlugin(ctx context.Context, config
 	})
 
 	return &numericalMedian{
-			offchainConfig,
-			onchainConfig,
-			fac.ContractTransmitter,
-			fac.DataSource,
-			fac.JuelsPerFeeCoinDataSource,
-			fac.GasPriceSubunitsDataSource,
-			fac.IncludeGasPriceSubunitsInObservation,
-			logger,
-			fac.ReportCodec,
-			deviationFunc,
-			fac.AcceptAfterFullTransmissionScheduleElapsed || offchainConfig.AcceptAfterFullTransmissionScheduleElapsed,
-
-			configuration.ConfigDigest,
-			configuration.F,
-			configuration.DurationAllTransmissionStages,
-			epochRound{},
-			new(big.Int),
-			time.Now(),
+		offchainConfig,
+		onchainConfig,
+		fac.ContractTransmitter,
+		fac.DataSource,
+		fac.JuelsPerFeeCoinDataSource,
+		fac.GasPriceSubunitsDataSource,
+		fac.IncludeGasPriceSubunitsInObservation,
+		logger,
+		fac.ReportCodec,
+		deviationFunc,
+		fac.AcceptAfterFullTransmissionScheduleElapsed || offchainConfig.AcceptAfterFullTransmissionScheduleElapsed,
+
+		configuration.ConfigDigest,
+		configuration.F,
+		configuration.DurationAllTransmissionStages,
+		epochRound{},
+		new(big.Int),
+		time.Now(),
+		maxReportLength,
+	}, types.ReportingPluginInfo{
+		"NumericalMedian",
+		false,
+		types.ReportingPluginLimits{
+			0,
+			maxObservationLength,
 			maxReportLength,
-		}, types.ReportingPluginInfo{
-			"NumericalMedian",
-			false,
-			types.ReportingPluginLimits{
-				0,
-				maxObservationLength,
-				maxReportLength,
-			},
-		}, nil
+		},
+	}, nil
 }
 
 func DefaultDeviationFunc(_ context.Context, thresholdPPB uint64, old *big.Int, new *big.Int) (bool, error) {

offchainreporting2/reportingplugin/titlerequest/titlerequest.go

@@ -40,24 +40,24 @@ type TitleRequestPluginFactory struct {
 
 func (fac *TitleRequestPluginFactory) NewReportingPlugin(_ context.Context, config types.ReportingPluginConfig) (types.ReportingPlugin, types.ReportingPluginInfo, error) {
 	return &TitleRequestPlugin{
-			config.F,
-			fac.Client,
-			fac.Contract,
-			map[[32]byte]bool{},
-			map[[32]byte]time.Time{},
-		}, types.ReportingPluginInfo{
-			"Title Request ReportingPlugin",
-			false,
-			types.ReportingPluginLimits{
-				// queries are empty
-				0,
-				// observations are at most 32 (request id) + 32 (title offset) + 32
-				// (title len) + maxTitleLen chars, let's generously round to 1000
-				1_000,
-				// reports follow the same format as observations
-				1_000,
-			},
-		}, nil
+		config.F,
+		fac.Client,
+		fac.Contract,
+		map[[32]byte]bool{},
+		map[[32]byte]time.Time{},
+	}, types.ReportingPluginInfo{
+		"Title Request ReportingPlugin",
+		false,
+		types.ReportingPluginLimits{
+			// queries are empty
+			0,
+			// observations are at most 32 (request id) + 32 (title offset) + 32
+			// (title len) + maxTitleLen chars, let's generously round to 1000
+			1_000,
+			// reports follow the same format as observations
+			1_000,
+		},
+	}, nil
 }
 
 var _ types.ReportingPlugin = (*TitleRequestPlugin)(nil)

offchainreporting2plus/internal/common/list/list.go

@@ -0,0 +1,61 @@
+package list
+
+import (
+	stdlist "container/list"
+	"iter"
+)
+
+// List is a type-safe wrapper around the standard library's container/list,
+// scoped to what we need: insertion at the back, removal of an arbitrary
+// element, and access to the front. All operations are O(1).
+type List[T any] struct {
+	internal stdlist.List
+}
+
+func NewList[T any]() *List[T] {
+	return &List[T]{}
+}
+
+// Element is a handle to an item in a List, needed to Remove it.
+type Element[T any] struct {
+	internal *stdlist.Element
+}
+
+func (e Element[T]) Value() T {
+	return e.internal.Value.(T)
+}
+
+func (l *List[T]) PushBack(item T) Element[T] {
+	return Element[T]{l.internal.PushBack(item)}
+}
+
+// Remove removes e from the list. Removing an element that has already been
+// removed is a no-op.
+func (l *List[T]) Remove(e Element[T]) {
+	l.internal.Remove(e.internal)
+}
+
+func (l *List[T]) Front() (T, bool) {
+	if e := l.internal.Front(); e != nil {
+		return e.Value.(T), true
+	} else {
+		var zero T
+		return zero, false
+	}
+}
+
+func (l *List[T]) Len() int {
+	return l.internal.Len()
+}
+
+// All iterates from front to back. The list must not be modified during
+// iteration.
+func (l *List[T]) All() iter.Seq[T] {
+	return func(yield func(T) bool) {
+		for e := l.internal.Front(); e != nil; e = e.Next() {
+			if !yield(e.Value.(T)) {
+				return
+			}
+		}
+	}
+}