Support sub-second reports

.changeset/empty-terms-mate.md

@@ -0,0 +1,9 @@
+---
+"chainlink": major
+---
+
+Upgrade LLO protocol to support sub-seconds reports.
+
+CAUTION: This release needs a careful rollout. It _must_ be tested on a staging DON first by setting half of the nodes to this version, while leaving half at the previous version, to verify inter-version interoperability before node-by-node rollout in production.
+
+NOTE: Protocol version 0 does NOT support gapless handover on sub-second reports. You must upgrade to version 1 for that.

core/services/llo/bm/dummy_transmitter.go

@@ -74,8 +74,8 @@ func (t *transmitter) Transmit(
 					"report.Report.ConfigDigest", r.ConfigDigest,
 					"report.Report.SeqNr", r.SeqNr,
 					"report.Report.ChannelID", r.ChannelID,
-					"report.Report.ValidAfterSeconds", r.ValidAfterSeconds,
-					"report.Report.ObservationTimestampSeconds", r.ObservationTimestampSeconds,
+					"report.Report.ValidAfterNanoseconds", r.ValidAfterNanoseconds,
+					"report.Report.ObservationTimestampNanoseconds", r.ObservationTimestampNanoseconds,
 					"report.Report.Values", r.Values,
 					"report.Report.Specimen", r.Specimen,
 				)

core/services/llo/evm/report_codec_common.go

@@ -0,0 +1,29 @@
+package evm
+
+import (
+	"fmt"
+	"math"
+
+	"github.com/smartcontractkit/chainlink-data-streams/llo"
+)
+
+// Extracts nanosecond timestamps as uint32 number of seconds
+func ExtractTimestamps(report llo.Report) (validAfterSeconds, observationTimestampSeconds uint32, err error) {
+	vas := report.ValidAfterNanoseconds / 1e9
+	ots := report.ObservationTimestampNanoseconds / 1e9
+	if vas > math.MaxUint32 {
+		err = fmt.Errorf("validAfterSeconds too large: %d", vas)
+		return
+	}
+	if ots > math.MaxUint32 {
+		err = fmt.Errorf("observationTimestampSeconds too large: %d", ots)
+		return
+	}
+	if vas >= ots {
+		// This should never happen, should have been prevented by the plugin,
+		// but we check anyway
+		err = fmt.Errorf("invariant violation: validAfterSeconds >= observationTimestampSeconds; got: %d >= %d", validAfterSeconds, observationTimestampSeconds)
+		return
+	}
+	return uint32(vas), uint32(ots), nil
+}

core/services/llo/evm/report_codec_evm_abi_encode_unpacked.go

@@ -103,13 +103,18 @@ func (r ReportCodecEVMABIEncodeUnpacked) Encode(ctx context.Context, report llo.
 		return nil, fmt.Errorf("failed to decode opts; got: '%s'; %w", cd.Opts, err)
 	}
 
+	validAfterSeconds, observationTimestampSeconds, err := ExtractTimestamps(report)
+	if err != nil {
+		return nil, fmt.Errorf("failed to extract timestamps; %w", err)
+	}
+
 	rf := BaseReportFields{
 		FeedID:             opts.FeedID,
-		ValidFromTimestamp: report.ValidAfterSeconds + 1,
-		Timestamp:          report.ObservationTimestampSeconds,
+		ValidFromTimestamp: validAfterSeconds + 1,
+		Timestamp:          observationTimestampSeconds,
 		NativeFee:          CalculateFee(nativePrice, opts.BaseUSDFee),
 		LinkFee:            CalculateFee(linkPrice, opts.BaseUSDFee),
-		ExpiresAt:          report.ObservationTimestampSeconds + opts.ExpirationWindow,
+		ExpiresAt:          observationTimestampSeconds + opts.ExpirationWindow,
 	}
 
 	header, err := r.buildHeader(ctx, rf)

core/services/llo/evm/report_codec_premium_legacy.go

@@ -88,21 +88,25 @@ func (r ReportCodecPremiumLegacy) Encode(ctx context.Context, report llo.Report,
 		multiplier = decimal.NewFromBigInt(opts.Multiplier.ToInt(), 0)
 	}
 
-	codec := reportcodecv3.NewReportCodec(opts.FeedID, r.Logger)
+	validAfterSeconds, observationTimestampSeconds, err := ExtractTimestamps(report)
+	if err != nil {
+		return nil, fmt.Errorf("failed to extract timestamps; %w", err)
+	}
 
 	rf := v3.ReportFields{
-		ValidFromTimestamp: report.ValidAfterSeconds + 1,
-		Timestamp:          report.ObservationTimestampSeconds,
+		ValidFromTimestamp: validAfterSeconds + 1,
+		Timestamp:          observationTimestampSeconds,
 		NativeFee:          CalculateFee(nativePrice, opts.BaseUSDFee),
 		LinkFee:            CalculateFee(linkPrice, opts.BaseUSDFee),
-		ExpiresAt:          report.ObservationTimestampSeconds + opts.ExpirationWindow,
+		ExpiresAt:          observationTimestampSeconds + opts.ExpirationWindow,
 		BenchmarkPrice:     quote.Benchmark.Mul(multiplier).BigInt(),
 		Bid:                quote.Bid.Mul(multiplier).BigInt(),
 		Ask:                quote.Ask.Mul(multiplier).BigInt(),
 	}
 
 	r.Logger.Debugw("Encoding report", "report", report, "opts", opts, "nativePrice", nativePrice, "linkPrice", linkPrice, "quote", quote, "multiplier", multiplier, "rf", rf)
 
+	codec := reportcodecv3.NewReportCodec(opts.FeedID, r.Logger)
 	return codec.BuildReport(ctx, rf)
 }
 

core/services/llo/retirement_report_orm.go

@@ -62,8 +62,12 @@ func (o *retirementReportCacheORM) LoadAttestedRetirementReports(ctx context.Con
 }
 
 func (o *retirementReportCacheORM) StoreConfig(ctx context.Context, cd ocr2types.ConfigDigest, signers [][]byte, f uint8) error {
-	_, err := o.ds.ExecContext(ctx, `INSERT INTO llo_retirement_report_cache_configs (config_digest, signers, f, updated_at) VALUES ($1, $2, $3, NOW())`, cd, signers, f)
-	return err
+	// do nothing on overwrite since configs are supposedly immutable
+	_, err := o.ds.ExecContext(ctx, `INSERT INTO llo_retirement_report_cache_configs (config_digest, signers, f, updated_at) VALUES ($1, $2, $3, NOW()) ON CONFLICT (config_digest) DO NOTHING`, cd, signers, f)
+	if err != nil {
+		return fmt.Errorf("StoreConfig failed: %w", err)
+	}
+	return nil
 }
 
 type Config struct {

core/services/llo/retirement_report_orm_test.go

@@ -42,6 +42,10 @@ func Test_RetirementReportCache_ORM(t *testing.T) {
 
 		err = orm.StoreConfig(ctx, cd2, signers, 2)
 		require.NoError(t, err)
+
+		// overwriting does nothing, the 255 is ignored
+		err = orm.StoreConfig(ctx, cd2, signers, 255)
+		require.NoError(t, err)
 	})
 	t.Run("LoadConfigs", func(t *testing.T) {
 		configs, err := orm.LoadConfigs(ctx)

core/services/ocr2/plugins/llo/helpers_test.go

@@ -252,6 +252,11 @@ func setupNode(
 		// [Log]
 		c.Log.Level = ptr(toml.LogLevel(zapcore.DebugLevel)) // generally speaking we want debug level for logs unless overridden
 
+		// [EVM.Transactions]
+		for _, evmCfg := range c.EVM {
+			evmCfg.Transactions.Enabled = ptr(false) // don't need txmgr
+		}
+
 		// Optional overrides
 		if f != nil {
 			f(c)
@@ -466,6 +471,16 @@ func createBridge(t *testing.T, bridgeName string, resultJSON string, borm bridg
 	}))
 }
 
+func addMemoStreamSpecs(t *testing.T, node Node, streams []Stream) {
+	for _, strm := range streams {
+		addStreamSpec(t, node, fmt.Sprintf("memo-%d", strm.id), &strm.id, fmt.Sprintf(`
+	value         [type=memo value="%s"];
+	multiply 	  [type=multiply times=1];
+	value -> multiply;
+	`, strm.baseBenchmarkPrice))
+	}
+}
+
 func addOCRJobsEVMPremiumLegacy(
 	t *testing.T,
 	streams []Stream,