llo: scaling improvements and safeguard channel definition version management

core/scripts/go.mod

@@ -373,7 +373,7 @@ require (
 	github.com/smartcontractkit/mcms v0.18.0 // indirect
 	github.com/smartcontractkit/tdh2/go/ocr2/decryptionplugin v0.0.0-20241009055228-33d0c0bf38de // indirect
 	github.com/smartcontractkit/tdh2/go/tdh2 v0.0.0-20241009055228-33d0c0bf38de // indirect
-	github.com/smartcontractkit/wsrpc v0.8.5-0.20250318131857-4568a0f8d12d // indirect
+	github.com/smartcontractkit/wsrpc v0.8.5-0.20250502134807-c57d3d995945 // indirect
 	github.com/sourcegraph/conc v0.3.0 // indirect
 	github.com/spf13/afero v1.12.0 // indirect
 	github.com/spf13/cast v1.7.1 // indirect

core/scripts/go.sum

@@ -1243,8 +1243,8 @@ github.com/smartcontractkit/tdh2/go/ocr2/decryptionplugin v0.0.0-20241009055228-
 github.com/smartcontractkit/tdh2/go/ocr2/decryptionplugin v0.0.0-20241009055228-33d0c0bf38de/go.mod h1:Sl2MF/Fp3fgJIVzhdGhmZZX2BlnM0oUUyBP4s4xYb6o=
 github.com/smartcontractkit/tdh2/go/tdh2 v0.0.0-20241009055228-33d0c0bf38de h1:66VQxXx3lvTaAZrMBkIcdH9VEjujUEvmBQdnyOJnkOc=
 github.com/smartcontractkit/tdh2/go/tdh2 v0.0.0-20241009055228-33d0c0bf38de/go.mod h1:NSc7hgOQbXG3DAwkOdWnZzLTZENXSwDJ7Va1nBp0YU0=
-github.com/smartcontractkit/wsrpc v0.8.5-0.20250318131857-4568a0f8d12d h1:B/LvtTIFwbkzoFE7723Q0IQBv/lcaTm0LgWBCZPUtvs=
-github.com/smartcontractkit/wsrpc v0.8.5-0.20250318131857-4568a0f8d12d/go.mod h1:m3pdp17i4bD50XgktkzWetcV5yaLsi7Gunbv4ZgN6qg=
+github.com/smartcontractkit/wsrpc v0.8.5-0.20250502134807-c57d3d995945 h1:zxcODLrFytOKmAd8ty8S/XK6WcIEJEgRBaL7sY/7l4Y=
+github.com/smartcontractkit/wsrpc v0.8.5-0.20250502134807-c57d3d995945/go.mod h1:m3pdp17i4bD50XgktkzWetcV5yaLsi7Gunbv4ZgN6qg=
 github.com/smarty/assertions v1.15.0/go.mod h1:yABtdzeQs6l1brC900WlRNwj6ZR55d7B+E8C6HtKdec=
 github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
 github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=

core/services/llo/channeldefinitions/onchain_channel_definition_cache.go

@@ -171,8 +171,10 @@ func (c *channelDefinitionCache) Start(ctx context.Context) error {
 			return err
 		} else if pd != nil {
 			c.definitions = pd.Definitions
-			c.initialBlockNum = pd.BlockNum + 1
 			c.definitionsVersion = uint32(pd.Version)
+			if pd.BlockNum+1 > c.initialBlockNum {
+				c.initialBlockNum = pd.BlockNum + 1
+			}
 		} else {
 			// ensure non-nil map ready for assignment later
 			c.definitions = make(llotypes.ChannelDefinitions)
@@ -264,6 +266,9 @@ func (c *channelDefinitionCache) readLogs(ctx context.Context) (err error) {
 		}
 		unpacked.DonId = new(big.Int).SetBytes(log.Topics[1])
 
+		//nolint:gosec // disable G115
+		unpacked.Raw.BlockNumber = uint64(log.BlockNumber)
+
 		if unpacked.DonId.Cmp(big.NewInt(int64(c.donID))) != 0 {
 			// skip logs for other donIDs, shouldn't happen given the
 			// FilterLogs call, but belts and braces
@@ -286,7 +291,8 @@ func (c *channelDefinitionCache) scanFromBlockNum() int64 {
 	c.newLogMu.RLock()
 	defer c.newLogMu.RUnlock()
 	if c.newLog != nil {
-		return int64(c.newLog.Raw.BlockNumber) + 1
+		//nolint:gosec // disable G115
+		return int64(c.newLog.Raw.BlockNumber)
 	}
 	return c.initialBlockNum
 }

core/services/llo/mercurytransmitter/orm.go

@@ -208,10 +208,10 @@ func (o *orm) Prune(ctx context.Context, serverURL string, maxSize, batchSize in
 		res, err = o.ds.ExecContext(ctx, `
 DELETE FROM llo_mercury_transmit_queue AS q
 USING (
-    SELECT transmission_hash 
+    SELECT transmission_hash
     FROM llo_mercury_transmit_queue
-    WHERE don_id = $1 
-      AND server_url = $2 
+    WHERE don_id = $1
+      AND server_url = $2
       AND seq_nr < $3
     ORDER BY seq_nr ASC
     LIMIT $4

core/services/llo/mercurytransmitter/queue.go

@@ -140,9 +140,16 @@ func (tq *transmitQueue) BlockingPop() (t *Transmission) {
 }
 
 func (tq *transmitQueue) IsEmpty() bool {
-	tq.mu.RLock()
-	defer tq.mu.RUnlock()
-	return tq.pq.Len() == 0
+	return tq.Len() == 0
+}
+
+func (tq *transmitQueue) Len() int {
+	tq.cond.L.Lock()
+	defer tq.cond.L.Unlock()
+
+	sz := tq.pq.Len()
+	tq.cond.Signal()
+	return sz
 }
 
 func (tq *transmitQueue) Start(context.Context) error {

core/services/llo/mercurytransmitter/queue_test.go

@@ -133,7 +133,7 @@ func Test_Queue(t *testing.T) {
 		}
 
 		tq.Push(testTransmissions[maxSize+3]) // push one more to trigger eviction
-		require.Equal(t, maxSize, tq.(*transmitQueue).pq.Len())
+		require.Equal(t, maxSize, tq.(*transmitQueue).Len())
 		require.Len(t, deleter.hashes, 4) // evicted overfill entries (3 oversize plus 1 more to make room)
 
 		// oldest entries removed

core/services/llo/mercurytransmitter/transmitter.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io"
 	"sync"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
@@ -29,7 +30,9 @@ import (
 
 const (
 	// Mercury server error codes
-	DuplicateReport = 2
+	DuplicateReport  = 2
+	commitInterval   = time.Millisecond * 25
+	commitBufferSize = 1000
 )
 
 var (
@@ -123,6 +126,8 @@ type transmitter struct {
 
 	stopCh services.StopChan
 	wg     *sync.WaitGroup
+
+	commitCh chan *Transmission
 }
 
 type Opts struct {
@@ -158,6 +163,7 @@ func newTransmitter(opts Opts) *transmitter {
 		opts.FromAccount,
 		make(services.StopChan),
 		&sync.WaitGroup{},
+		make(chan *Transmission, 1000*len(servers)),
 	}
 }
 
@@ -200,6 +206,7 @@ func (mt *transmitter) Start(ctx context.Context) (err error) {
 			})
 		}
 
+		mt.spawnCommitLoops()
 		return g.Wait()
 	})
 }
@@ -248,37 +255,36 @@ func (mt *transmitter) Transmit(
 	sigs []types.AttributedOnchainSignature,
 ) (err error) {
 	ok := mt.IfStarted(func() {
-		err = mt.transmit(ctx, digest, seqNr, report, sigs)
+		for serverURL := range mt.servers {
+			t := &Transmission{
+				ServerURL:    serverURL,
+				ConfigDigest: digest,
+				SeqNr:        seqNr,
+				Report:       report,
+				Sigs:         sigs,
+			}
+			select {
+			case mt.commitCh <- t:
+			case <-ctx.Done():
+				err = fmt.Errorf("failed to add transmission to commit channel: %w", ctx.Err())
+			}
+		}
 	})
+
 	if !ok {
 		return errors.New("transmitter is not started")
 	}
-	return
+
+	return err
 }
 
-func (mt *transmitter) transmit(
-	ctx context.Context,
-	digest types.ConfigDigest,
-	seqNr uint64,
-	report ocr3types.ReportWithInfo[llotypes.ReportInfo],
-	sigs []types.AttributedOnchainSignature,
-) error {
+func (mt *transmitter) transmit(ctx context.Context, transmissions []*Transmission) error {
 	// On shutdown appears that libocr can pass us a pre-canceled context;
 	// don't even bother trying to insert/transmit in this case
 	if ctx.Err() != nil {
 		return fmt.Errorf("cannot transmit; context already canceled: %w", ctx.Err())
 	}
 
-	transmissions := make([]*Transmission, 0, len(mt.servers))
-	for serverURL := range mt.servers {
-		transmissions = append(transmissions, &Transmission{
-			ServerURL:    serverURL,
-			ConfigDigest: digest,
-			SeqNr:        seqNr,
-			Report:       report,
-			Sigs:         sigs,
-		})
-	}
 	// NOTE: This insert on its own can leave orphaned records in the case of
 	// shutdown, because:
 	// 1. Transmitter is shut down after oracle
@@ -312,11 +318,15 @@ func (mt *transmitter) transmit(
 	for i := range transmissions {
 		t := transmissions[i]
 		if mt.verboseLogging {
-			mt.lggr.Debugw("Transmit report", "digest", digest.Hex(), "seqNr", seqNr, "reportFormat", report.Info.ReportFormat, "reportLifeCycleStage", report.Info.LifeCycleStage, "transmissionHash", fmt.Sprintf("%x", t.Hash()))
+			mt.lggr.Debugw("Transmit report",
+				"digest", t.ConfigDigest.Hex(), "seqNr", t.SeqNr, "reportFormat", t.Report.Info.ReportFormat,
+				"reportLifeCycleStage", t.Report.Info.LifeCycleStage,
+				"transmissionHash", fmt.Sprintf("%x", t.Hash()))
 		}
-		s := mt.servers[t.ServerURL]
+
 		// OK to do this synchronously since pushing to queue is just a mutex
 		// lock and array append and ought to be extremely fast
+		s := mt.servers[t.ServerURL]
 		if ok := s.q.Push(t); !ok {
 			s.transmitQueuePushErrorCount.Inc()
 			// This shouldn't be possible since transmitter is always shut down
@@ -332,3 +342,53 @@ func (mt *transmitter) transmit(
 func (mt *transmitter) FromAccount(ctx context.Context) (ocrtypes.Account, error) {
 	return ocrtypes.Account(mt.fromAccount), nil
 }
+
+func (mt *transmitter) spawnCommitLoops() {
+	for x := 0; x < len(mt.servers); x++ {
+		mt.wg.Add(1)
+
+		go func() {
+			defer mt.wg.Done()
+
+			var err error
+			ctx, cancel := mt.stopCh.NewCtx()
+			defer cancel()
+
+			buff := cap(mt.commitCh) / 10
+			transmissions := make([]*Transmission, 0, buff)
+			ticker := time.NewTicker(commitInterval)
+			defer ticker.Stop()
+
+			for {
+				select {
+				case <-ctx.Done():
+					if len(transmissions) >= buff {
+						closeCtx, closeCancel := context.WithTimeout(context.Background(), time.Second)
+						defer closeCancel()
+						if err = mt.transmit(closeCtx, transmissions); err != nil {
+							mt.lggr.Error("Error transmitting records when stopping", "error", err)
+						}
+					}
+					return
+
+				case <-ticker.C:
+					if len(transmissions) > 0 {
+						err = mt.transmit(ctx, transmissions)
+						transmissions = make([]*Transmission, 0, buff)
+					}
+
+				case t := <-mt.commitCh:
+					transmissions = append(transmissions, t)
+					if len(transmissions) >= buff {
+						err = mt.transmit(ctx, transmissions)
+						transmissions = make([]*Transmission, 0, buff)
+					}
+				}
+
+				if err != nil {
+					mt.lggr.Error("Error transmitting records", "error", err)
+				}
+			}
+		}()
+	}
+}

core/services/llo/mercurytransmitter/transmitter_test.go

@@ -110,6 +110,7 @@ func Test_Transmitter_Transmit(t *testing.T) {
 				require.NoError(t, mt.servers[sURL].q.Init([]*Transmission{}))
 				require.NoError(t, mt.servers[sURL2].q.Init([]*Transmission{}))
 				require.NoError(t, mt.servers[sURL3].q.Init([]*Transmission{}))
+				mt.spawnCommitLoops()
 
 				return nil
 			})
@@ -125,24 +126,27 @@ func Test_Transmitter_Transmit(t *testing.T) {
 			err = mt.Transmit(testutils.Context(t), digest, seqNr, report, sigs)
 			require.NoError(t, err)
 
+			// wait for the commit loop to run
+			time.Sleep(2 * commitInterval)
+
 			// ensure it was added to the queue
-			require.Equal(t, 1, mt.servers[sURL].q.(*transmitQueue).pq.Len())
+			require.Equal(t, 1, mt.servers[sURL].q.(*transmitQueue).Len())
 			assert.Equal(t, &Transmission{
 				ServerURL:    sURL,
 				ConfigDigest: digest,
 				SeqNr:        seqNr,
 				Report:       report,
 				Sigs:         sigs,
 			}, mt.servers[sURL].q.(*transmitQueue).pq.Pop().(*Transmission))
-			require.Equal(t, 1, mt.servers[sURL2].q.(*transmitQueue).pq.Len())
+			require.Equal(t, 1, mt.servers[sURL2].q.(*transmitQueue).Len())
 			assert.Equal(t, &Transmission{
 				ServerURL:    sURL2,
 				ConfigDigest: digest,
 				SeqNr:        seqNr,
 				Report:       report,
 				Sigs:         sigs,
 			}, mt.servers[sURL2].q.(*transmitQueue).pq.Pop().(*Transmission))
-			require.Equal(t, 1, mt.servers[sURL3].q.(*transmitQueue).pq.Len())
+			require.Equal(t, 1, mt.servers[sURL3].q.(*transmitQueue).Len())
 			assert.Equal(t, &Transmission{
 				ServerURL:    sURL3,
 				ConfigDigest: digest,