diff --git a/pkg/settings/cresettings/README.md b/pkg/settings/cresettings/README.md index 87120257d2..1e1b554f0b 100644 --- a/pkg/settings/cresettings/README.md +++ b/pkg/settings/cresettings/README.md @@ -373,6 +373,9 @@ flowchart %% the trigger coordinator's reader is likewise its own entry point CentralTriggerQueue.Put +%% TODO placating test for now since this flowchart no longer renders + DonTimeSequencedTimestampsActivePeriod + classDef bound stroke:#f00 classDef gate stroke:#0f0 classDef queue stroke:#00f diff --git a/pkg/settings/cresettings/defaults.json b/pkg/settings/cresettings/defaults.json index e047845f2e..8fe7160a9e 100644 --- a/pkg/settings/cresettings/defaults.json +++ b/pkg/settings/cresettings/defaults.json @@ -59,6 +59,7 @@ "VaultMaxPerOracleUnexpiredBlobCumulativePayloadSizeLimit": "31.45728mb", "VaultMaxPerOracleUnexpiredBlobCount": "1000", "MissingRequestRecoveryEnabled": "false", + "DonTimeSequencedTimestampsActivePeriod": "[2100-01-01 00:00:00 +0000 UTC,2101-01-01 00:00:00 +0000 UTC]", "ConfidentialCompute": { "GlobalRate": "1000rps:1000", "MaxRetries": "3", diff --git a/pkg/settings/cresettings/defaults.toml b/pkg/settings/cresettings/defaults.toml index 7311ee04e6..832b3df2ca 100644 --- a/pkg/settings/cresettings/defaults.toml +++ b/pkg/settings/cresettings/defaults.toml @@ -58,6 +58,7 @@ VaultMaxBlobPayloadSizeLimit = '25.6kb' VaultMaxPerOracleUnexpiredBlobCumulativePayloadSizeLimit = '31.45728mb' VaultMaxPerOracleUnexpiredBlobCount = '1000' MissingRequestRecoveryEnabled = 'false' +DonTimeSequencedTimestampsActivePeriod = '[2100-01-01 00:00:00 +0000 UTC,2101-01-01 00:00:00 +0000 UTC]' [ConfidentialCompute] GlobalRate = '1000rps:1000' diff --git a/pkg/settings/cresettings/settings.go b/pkg/settings/cresettings/settings.go index 458253b058..4477d4ffa1 100644 --- a/pkg/settings/cresettings/settings.go +++ b/pkg/settings/cresettings/settings.go @@ -51,6 +51,12 @@ var DefaultGetter Getter // Deprecated: use Default var Config Schema +var ( + year2100 = time.Date(2100, 1, 1, 0, 0, 0, 0, time.UTC) + year2101 = time.Date(2101, 1, 1, 0, 0, 0, 0, time.UTC) + disabledFeatureTimeRange = TimeRange(year2100, year2101) +) + var Default = Schema{ WorkflowLimit: Int(1000), WorkflowExecutionConcurrencyLimit: Int(1000), @@ -160,6 +166,8 @@ var Default = Schema{ // MissingRequestRecoveryEnabled MissingRequestRecoveryEnabled: Bool(false), + DonTimeSequencedTimestampsActivePeriod: disabledFeatureTimeRange, + // Confidential Compute (San Marino framework) node-level settings. Defaults // mirror the previous hardcoded executor defaults so behavior is unchanged // until explicitly overridden. @@ -335,38 +343,23 @@ var Default = Schema{ RequestTimeout: Duration(30 * time.Second), }, - FeatureHTTPTriggerNewExecutionIDsActivePeriod: TimeRange( - time.Date(2100, 1, 1, 0, 0, 0, 0, time.UTC), - time.Date(2101, 1, 1, 0, 0, 0, 0, time.UTC)), - FeatureChainCapabilityHashBasedOCRActivePeriod: TimeRange( - time.Date(2100, 1, 1, 0, 0, 0, 0, time.UTC), - time.Date(2101, 1, 1, 0, 0, 0, 0, time.UTC)), - FeatureEVMWriteReportL1FeeActivePeriod: TimeRange( - time.Date(2100, 1, 1, 0, 0, 0, 0, time.UTC), - time.Date(2101, 1, 1, 0, 0, 0, 0, time.UTC)), - FeatureAptosWriteReportBlockTimestampActivePeriod: TimeRange( - time.Date(2100, 1, 1, 0, 0, 0, 0, time.UTC), - time.Date(2101, 1, 1, 0, 0, 0, 0, time.UTC)), + FeatureHTTPTriggerNewExecutionIDsActivePeriod: disabledFeatureTimeRange, + FeatureChainCapabilityHashBasedOCRActivePeriod: disabledFeatureTimeRange, + FeatureEVMWriteReportL1FeeActivePeriod: disabledFeatureTimeRange, + FeatureAptosWriteReportBlockTimestampActivePeriod: disabledFeatureTimeRange, // ON by default: covers all possible timestamps including zero time.Time{}, // so WorkflowTag is included in the hash matching current prod behavior. // After rollout, set to far-future window to exclude WorkflowTag. FeatureRequestHashIncludeWorkflowTagActivePeriod: TimeRange( - time.Date(1, 1, 1, 0, 0, 0, 0, time.UTC), - time.Date(2100, 1, 1, 0, 0, 0, 0, time.UTC)), + time.Date(1, 1, 1, 0, 0, 0, 0, time.UTC), year2100), // OFF by default: the workflow_specs_v2.workflow_tag reconcile backfill // is intentionally disabled on a fresh deploy. Ops narrows the range to // cover "now" only after FeatureRequestHashIncludeWorkflowTag is muted // on every DON member, so DBs can heal without producing tag-driven // hash divergence during the fill window. - FeatureWorkflowTagBackfillActivePeriod: TimeRange( - time.Date(2100, 1, 1, 0, 0, 0, 0, time.UTC), - time.Date(2101, 1, 1, 0, 0, 0, 0, time.UTC)), - FeatureConsensusStricterMedianQuorumActivePeriod: TimeRange( - time.Date(2100, 1, 1, 0, 0, 0, 0, time.UTC), - time.Date(2101, 1, 1, 0, 0, 0, 0, time.UTC)), - FeatureConsensusIncludeAllTimestampsActivePeriod: TimeRange( - time.Date(2100, 1, 1, 0, 0, 0, 0, time.UTC), - time.Date(2101, 1, 1, 0, 0, 0, 0, time.UTC)), + FeatureWorkflowTagBackfillActivePeriod: disabledFeatureTimeRange, + FeatureConsensusStricterMedianQuorumActivePeriod: disabledFeatureTimeRange, + FeatureConsensusIncludeAllTimestampsActivePeriod: disabledFeatureTimeRange, }, } @@ -481,6 +474,8 @@ type Schema struct { MissingRequestRecoveryEnabled Setting[bool] + DonTimeSequencedTimestampsActivePeriod Setting[Range[config.Timestamp]] + // Confidential Compute (San Marino framework) node-level settings. ConfidentialCompute confidentialCompute diff --git a/pkg/settings/limits/range.go b/pkg/settings/limits/range.go index a77e77f008..a21fe9116c 100644 --- a/pkg/settings/limits/range.go +++ b/pkg/settings/limits/range.go @@ -15,14 +15,14 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/settings" ) -// BoundLimiter is a limiter for simple bounds checks. +// RangeLimiter is a limiter for bounded range checks. type RangeLimiter[N Number] interface { Limiter[settings.Range[N]] - // Check returns ErrorBoundLimited if the value is above the limit. + // Check returns ErrorRangeLimited if the value is above the limit. Check(context.Context, N) error } -// NewRangeLimiter returns a RangeLimiter with the given lower bounds. +// NewRangeLimiter returns a RangeLimiter with the given bounds. func NewRangeLimiter[N Number](bounds settings.Range[N]) RangeLimiter[N] { return &simpleRangeLimiter[N]{bounds: bounds} } diff --git a/pkg/workflows/dontime/factory.go b/pkg/workflows/dontime/factory.go index 76c102547a..93e973b846 100644 --- a/pkg/workflows/dontime/factory.go +++ b/pkg/workflows/dontime/factory.go @@ -9,8 +9,11 @@ import ( "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" + "github.com/smartcontractkit/chainlink-common/pkg/config" "github.com/smartcontractkit/chainlink-common/pkg/logger" "github.com/smartcontractkit/chainlink-common/pkg/services" + "github.com/smartcontractkit/chainlink-common/pkg/settings/cresettings" + "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" "github.com/smartcontractkit/chainlink-common/pkg/types/core" "github.com/smartcontractkit/chainlink-common/pkg/workflows/dontime/pb" ) @@ -26,19 +29,30 @@ const ( var _ core.OCR3ReportingPluginFactory = &Factory{} type Factory struct { - store *Store - lggr logger.Logger + store *Store + lggr logger.Logger + sequencedTSEnabled limits.RangeLimiter[config.Timestamp] services.StateMachine } func NewFactory(s *Store, lggr logger.Logger) (*Factory, error) { return &Factory{ - store: s, - lggr: logger.Named(lggr, "OCR3DonTimeFactory"), + store: s, + lggr: logger.Named(lggr, "OCR3DonTimeFactory"), + sequencedTSEnabled: limits.NewRangeLimiter(cresettings.Default.DonTimeSequencedTimestampsActivePeriod.DefaultValue), }, nil } +func (o *Factory) InitLimits(lf limits.Factory) error { + sequencedTSEnabled, err := limits.MakeRangeLimiter[config.Timestamp](lf, cresettings.Default.DonTimeSequencedTimestampsActivePeriod) + if err != nil { + return err + } + o.sequencedTSEnabled = sequencedTSEnabled + return nil +} + func (o *Factory) NewReportingPlugin(_ context.Context, config ocr3types.ReportingPluginConfig) (ocr3types.ReportingPlugin[[]byte], ocr3types.ReportingPluginInfo, error) { var configProto pb.Config err := proto.Unmarshal(config.OffchainConfig, &configProto) @@ -75,6 +89,7 @@ func (o *Factory) NewReportingPlugin(_ context.Context, config ocr3types.Reporti if err != nil { return nil, ocr3types.ReportingPluginInfo{}, err } + plugin.setSequencedTSEnabled(o.sequencedTSEnabled) pluginInfo := ocr3types.ReportingPluginInfo{ Name: "DON Time Plugin", Limits: ocr3types.ReportingPluginLimits{ diff --git a/pkg/workflows/dontime/pb/dontime.go b/pkg/workflows/dontime/pb/dontime.go new file mode 100644 index 0000000000..87e2b4deaf --- /dev/null +++ b/pkg/workflows/dontime/pb/dontime.go @@ -0,0 +1,22 @@ +package pb + +// MaxSeqNum returns the max sequence number from TimestampsBySequence, or -1 if none exist. +func (t *ObservedDonTimes) MaxSeqNum() int64 { + var maxSeqNum int64 = -1 + for seqNum := range t.TimestampsBySequence { + if seqNum > maxSeqNum { + maxSeqNum = seqNum + } + } + return maxSeqNum +} + +// EarliestTS returns the earliest timestamp value from TimestampsBySequence or nil if none exist. +func (t *ObservedDonTimes) EarliestTS() (earliestTS *int64) { + for _, ts := range t.TimestampsBySequence { + if earliestTS == nil || ts < *earliestTS { + earliestTS = &ts + } + } + return +} diff --git a/pkg/workflows/dontime/pb/dontime.pb.go b/pkg/workflows/dontime/pb/dontime.pb.go index 36d662710f..961d02094e 100644 --- a/pkg/workflows/dontime/pb/dontime.pb.go +++ b/pkg/workflows/dontime/pb/dontime.pb.go @@ -82,10 +82,12 @@ func (x *Observation) GetLimitByBatchSizeFlag() bool { } type ObservedDonTimes struct { - state protoimpl.MessageState `protogen:"open.v1"` - Timestamps []int64 `protobuf:"varint,1,rep,packed,name=timestamps,proto3" json:"timestamps,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + state protoimpl.MessageState `protogen:"open.v1"` + // Deprecated: Marked as deprecated in dontime.proto. + Timestamps []int64 `protobuf:"varint,1,rep,packed,name=timestamps,proto3" json:"timestamps,omitempty"` + TimestampsBySequence map[int64]int64 `protobuf:"bytes,2,rep,name=timestampsBySequence,proto3" json:"timestampsBySequence,omitempty" protobuf_key:"varint,1,opt,name=key" protobuf_val:"varint,2,opt,name=value"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *ObservedDonTimes) Reset() { @@ -118,6 +120,7 @@ func (*ObservedDonTimes) Descriptor() ([]byte, []int) { return file_dontime_proto_rawDescGZIP(), []int{1} } +// Deprecated: Marked as deprecated in dontime.proto. func (x *ObservedDonTimes) GetTimestamps() []int64 { if x != nil { return x.Timestamps @@ -125,6 +128,13 @@ func (x *ObservedDonTimes) GetTimestamps() []int64 { return nil } +func (x *ObservedDonTimes) GetTimestampsBySequence() map[int64]int64 { + if x != nil { + return x.TimestampsBySequence + } + return nil +} + type Outcome struct { state protoimpl.MessageState `protogen:"open.v1"` Timestamp int64 `protobuf:"varint,1,opt,name=timestamp,proto3" json:"timestamp,omitempty"` @@ -188,11 +198,15 @@ const file_dontime_proto_rawDesc = "" + "\x18limit_by_batch_size_flag\x18\x04 \x01(\bR\x14limitByBatchSizeFlag\x1a;\n" + "\rRequestsEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + - "\x05value\x18\x02 \x01(\x03R\x05value:\x028\x01J\x04\b\x03\x10\x04\"2\n" + - "\x10ObservedDonTimes\x12\x1e\n" + + "\x05value\x18\x02 \x01(\x03R\x05value:\x028\x01J\x04\b\x03\x10\x04\"\xe0\x01\n" + + "\x10ObservedDonTimes\x12\"\n" + "\n" + - "timestamps\x18\x01 \x03(\x03R\n" + - "timestamps\"\xcd\x01\n" + + "timestamps\x18\x01 \x03(\x03B\x02\x18\x01R\n" + + "timestamps\x12_\n" + + "\x14timestampsBySequence\x18\x02 \x03(\v2+.ObservedDonTimes.TimestampsBySequenceEntryR\x14timestampsBySequence\x1aG\n" + + "\x19TimestampsBySequenceEntry\x12\x10\n" + + "\x03key\x18\x01 \x01(\x03R\x03key\x12\x14\n" + + "\x05value\x18\x02 \x01(\x03R\x05value:\x028\x01\"\xcd\x01\n" + "\aOutcome\x12\x1c\n" + "\ttimestamp\x18\x01 \x01(\x03R\ttimestamp\x12L\n" + "\x12observed_don_times\x18\x02 \x03(\v2\x1e.Outcome.ObservedDonTimesEntryR\x10observedDonTimes\x1aV\n" + @@ -212,23 +226,25 @@ func file_dontime_proto_rawDescGZIP() []byte { return file_dontime_proto_rawDescData } -var file_dontime_proto_msgTypes = make([]protoimpl.MessageInfo, 5) +var file_dontime_proto_msgTypes = make([]protoimpl.MessageInfo, 6) var file_dontime_proto_goTypes = []any{ (*Observation)(nil), // 0: Observation (*ObservedDonTimes)(nil), // 1: ObservedDonTimes (*Outcome)(nil), // 2: Outcome nil, // 3: Observation.RequestsEntry - nil, // 4: Outcome.ObservedDonTimesEntry + nil, // 4: ObservedDonTimes.TimestampsBySequenceEntry + nil, // 5: Outcome.ObservedDonTimesEntry } var file_dontime_proto_depIdxs = []int32{ 3, // 0: Observation.requests:type_name -> Observation.RequestsEntry - 4, // 1: Outcome.observed_don_times:type_name -> Outcome.ObservedDonTimesEntry - 1, // 2: Outcome.ObservedDonTimesEntry.value:type_name -> ObservedDonTimes - 3, // [3:3] is the sub-list for method output_type - 3, // [3:3] is the sub-list for method input_type - 3, // [3:3] is the sub-list for extension type_name - 3, // [3:3] is the sub-list for extension extendee - 0, // [0:3] is the sub-list for field type_name + 4, // 1: ObservedDonTimes.timestampsBySequence:type_name -> ObservedDonTimes.TimestampsBySequenceEntry + 5, // 2: Outcome.observed_don_times:type_name -> Outcome.ObservedDonTimesEntry + 1, // 3: Outcome.ObservedDonTimesEntry.value:type_name -> ObservedDonTimes + 4, // [4:4] is the sub-list for method output_type + 4, // [4:4] is the sub-list for method input_type + 4, // [4:4] is the sub-list for extension type_name + 4, // [4:4] is the sub-list for extension extendee + 0, // [0:4] is the sub-list for field type_name } func init() { file_dontime_proto_init() } @@ -242,7 +258,7 @@ func file_dontime_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_dontime_proto_rawDesc), len(file_dontime_proto_rawDesc)), NumEnums: 0, - NumMessages: 5, + NumMessages: 6, NumExtensions: 0, NumServices: 0, }, diff --git a/pkg/workflows/dontime/pb/dontime.proto b/pkg/workflows/dontime/pb/dontime.proto index d0baecf3b3..ca82d41d76 100644 --- a/pkg/workflows/dontime/pb/dontime.proto +++ b/pkg/workflows/dontime/pb/dontime.proto @@ -10,7 +10,8 @@ message Observation { } message ObservedDonTimes { - repeated int64 timestamps = 1; + repeated int64 timestamps = 1 [deprecated = true]; + map timestampsBySequence = 2; } message Outcome { diff --git a/pkg/workflows/dontime/plugin.go b/pkg/workflows/dontime/plugin.go index 5c91cbbfb9..74403dde1c 100644 --- a/pkg/workflows/dontime/plugin.go +++ b/pkg/workflows/dontime/plugin.go @@ -5,6 +5,7 @@ import ( "context" "errors" "fmt" + "maps" "slices" "time" @@ -18,7 +19,10 @@ import ( "github.com/smartcontractkit/libocr/quorumhelper" "github.com/smartcontractkit/chainlink-common/pkg/beholder" + "github.com/smartcontractkit/chainlink-common/pkg/config" "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/settings/cresettings" + "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" "github.com/smartcontractkit/chainlink-common/pkg/workflows/dontime/pb" ) @@ -92,6 +96,8 @@ type Plugin struct { minTimeIncrease int64 metrics pluginMetrics + + sequencedTSEnabled limits.RangeLimiter[config.Timestamp] } var _ ocr3types.ReportingPlugin[[]byte] = (*Plugin)(nil) @@ -113,16 +119,21 @@ func NewPlugin(store *Store, config ocr3types.ReportingPluginConfig, offchainCfg } return &Plugin{ - store: store, - config: config, - offChainConfig: offchainCfg, - lggr: logger.Named(lggr, "DONTimePlugin"), - batchSize: int(offchainCfg.MaxBatchSize), - minTimeIncrease: offchainCfg.MinTimeIncrease / int64(time.Millisecond), - metrics: metrics, + store: store, + config: config, + offChainConfig: offchainCfg, + lggr: logger.Named(lggr, "DONTimePlugin"), + batchSize: int(offchainCfg.MaxBatchSize), + minTimeIncrease: offchainCfg.MinTimeIncrease / int64(time.Millisecond), + metrics: metrics, + sequencedTSEnabled: limits.NewRangeLimiter(cresettings.Default.DonTimeSequencedTimestampsActivePeriod.DefaultValue), }, nil } +func (p *Plugin) setSequencedTSEnabled(enabledRange limits.RangeLimiter[config.Timestamp]) { + p.sequencedTSEnabled = enabledRange +} + func (p *Plugin) Query(_ context.Context, _ ocr3types.OutcomeContext) (types.Query, error) { return nil, nil } @@ -132,11 +143,7 @@ func sortedRequests(requests map[string]*Request) []*Request { return nil } - ids := make([]string, 0, len(requests)) - for id := range requests { - ids = append(ids, id) - } - slices.Sort(ids) + ids := slices.Sorted(maps.Keys(requests)) sorted := make([]*Request, 0, len(ids)) for _, id := range ids { @@ -146,48 +153,20 @@ func sortedRequests(requests map[string]*Request) []*Request { } func (p *Plugin) Observation(ctx context.Context, outctx ocr3types.OutcomeContext, query types.Query) (types.Observation, error) { - previousOutcome := &pb.Outcome{} - if err := proto.Unmarshal(outctx.PreviousOutcome, previousOutcome); err != nil { - p.lggr.Errorf("failed to unmarshal previous outcome in Observation phase") - } - sortedRequests := sortedRequests(p.store.GetRequests()) requests := map[string]int64{} // Maps executionID --> seqNum - removedCount := 0 for _, req := range sortedRequests { - // Validate request sequence number - numObservedDonTimes := 0 - times, ok := previousOutcome.ObservedDonTimes[req.WorkflowExecutionID] - if ok { - // We have seen this workflow before so check against the sequence - numObservedDonTimes = len(times.Timestamps) - } - - if req.SeqNum > numObservedDonTimes { - p.store.RemoveRequest(req.WorkflowExecutionID) - req.SendResponse(Response{ - WorkflowExecutionID: req.WorkflowExecutionID, - SeqNum: req.SeqNum, - Timestamp: 0, - Err: fmt.Errorf("requested seqNum %d for executionID %s is greater than the number of observed don times %d", - req.SeqNum, req.WorkflowExecutionID, numObservedDonTimes), - }) - removedCount += 1 - continue - } - requests[req.WorkflowExecutionID] = int64(req.SeqNum) if len(requests) >= p.batchSize { break } } - overflowCount := len(sortedRequests) - len(requests) - removedCount + overflowCount := len(sortedRequests) - len(requests) p.lggr.Debugw("Observation batch processed", "inputRequests", len(sortedRequests), "batchSize", p.batchSize, "includedRequests", len(requests), - "removedRequests", removedCount, "overflowRequests", overflowCount, ) if overflowCount > 0 { @@ -213,22 +192,12 @@ func (p *Plugin) ObservationQuorum(_ context.Context, _ ocr3types.OutcomeContext } func (p *Plugin) Outcome(ctx context.Context, outctx ocr3types.OutcomeContext, _ types.Query, aos []types.AttributedObservation) (ocr3types.Outcome, error) { - observationCounts := map[string]int64{} // counts how many nodes reported where a new DON timestamp might be needed type timestampNodePair struct { Timestamp int64 NodeID int OffsetFromMedian int64 } var timestampNodePairs []timestampNodePair - - prevOutcome := &pb.Outcome{} - if err := proto.Unmarshal(outctx.PreviousOutcome, prevOutcome); err != nil { - p.lggr.Errorf("failed to unmarshal previous outcome in Outcome phase") - } - if prevOutcome.ObservedDonTimes == nil { - prevOutcome.ObservedDonTimes = make(map[string]*pb.ObservedDonTimes) - } - for idx, ao := range aos { observation := &pb.Observation{} if err := proto.Unmarshal(ao.Observation, observation); err != nil { @@ -236,23 +205,9 @@ func (p *Plugin) Outcome(ctx context.Context, outctx ocr3types.OutcomeContext, _ continue } - for id, requestSeqNum := range observation.Requests { - var currSeqNum int64 - if times, ok := prevOutcome.ObservedDonTimes[id]; ok { - currSeqNum = int64(len(times.Timestamps)) - } - // We only count requests for the next sequence number and ignore all other ones. - if requestSeqNum == currSeqNum { - observationCounts[id]++ - } else if requestSeqNum > currSeqNum { - // This should never happen since we don't include out of sequence requests in the Observation phase - p.lggr.Errorf("request seqNum %d for executionID %s is greater than the number of observed don times %d", - requestSeqNum, id, currSeqNum) - } - } - timestampNodePairs = append(timestampNodePairs, timestampNodePair{Timestamp: observation.Timestamp, NodeID: idx}) } + if len(timestampNodePairs) == 0 { return nil, errors.New("no observation contains a valid timestamp") } @@ -272,15 +227,96 @@ func (p *Plugin) Outcome(ctx context.Context, outctx ocr3types.OutcomeContext, _ "maxOffsetFromMedian", timestampNodePairs[len(timestampNodePairs)-1].OffsetFromMedian, ) - outcome := prevOutcome + prevOutcome := &pb.Outcome{} + if err := proto.Unmarshal(outctx.PreviousOutcome, prevOutcome); err != nil { + p.lggr.Errorf("failed to unmarshal previous outcome in Outcome phase") + } + if prevOutcome.ObservedDonTimes == nil { + prevOutcome.ObservedDonTimes = make(map[string]*pb.ObservedDonTimes) + } // Compare with prior outcome to ensure DON time never goes backward. - if donTime < outcome.Timestamp+p.minTimeIncrease { + if donTime < prevOutcome.Timestamp+p.minTimeIncrease { p.lggr.Infow("DON Time incremented by minimum time increase to ensure time progression", "minTimeIncrease", p.minTimeIncrease) - donTime = outcome.Timestamp + p.minTimeIncrease + donTime = prevOutcome.Timestamp + p.minTimeIncrease } p.lggr.Infow("New DON Time", "donTime", donTime) + + var outcome *pb.Outcome + if err := p.sequencedTSEnabled.Check(ctx, config.NewTimestamp(time.UnixMilli(donTime))); err != nil { + if !errors.Is(err, limits.ErrorRangeLimited[config.Timestamp]{}) { + p.lggr.Warnw("Failed to check for sequenced timestamp feature flag", "err", err) + } + outcome = p.unsequencedOutcome(aos, prevOutcome, donTime) + } else { + outcome = p.sequencedOutcome(aos, prevOutcome, donTime) + } + + var outcomeBatchOverflowCount int64 + if len(outcome.ObservedDonTimes) > p.batchSize { + ids := slices.Sorted(maps.Keys(outcome.ObservedDonTimes)) + outcomeBatchOverflowCount = int64(len(ids) - p.batchSize) + for _, id := range ids[p.batchSize:] { + delete(outcome.ObservedDonTimes, id) + } + p.lggr.Warnw("Trimmed outcome observed don times to batch size", + "batchSize", p.batchSize, + "removedEntries", outcomeBatchOverflowCount, + ) + } + + outcomeBytes, err := proto.MarshalOptions{Deterministic: true}.Marshal(outcome) + p.lggr.Infow("Outcome computed", + "observedDonTimesEntries", len(outcome.ObservedDonTimes), + "outcomeSizeBytes", len(outcomeBytes), + ) + p.metrics.donTime.Record(ctx, outcome.Timestamp) + p.metrics.donTimeEntries.Record(ctx, int64(len(outcome.ObservedDonTimes))) + p.metrics.outcomeBatchOverflow.Record(ctx, outcomeBatchOverflowCount) + p.metrics.outcomeSize.Record(ctx, int64(len(outcomeBytes))) + return outcomeBytes, err +} + +// unsequencedOutcome executes the original outcome logic to produce an unsequenced slice of [pb.ObservedDonTimes.Timestamps]. +func (p *Plugin) unsequencedOutcome(aos []types.AttributedObservation, prevOutcome *pb.Outcome, donTime int64) *pb.Outcome { + // If disabling the feature flag, then at the transition point, we need to convert from the map format to the slices + for _, observedTimes := range prevOutcome.ObservedDonTimes { + if mapLen := len(observedTimes.TimestampsBySequence); mapLen > 0 { + observedTimes.Timestamps = make([]int64, mapLen) + for seqNum, ts := range observedTimes.TimestampsBySequence { + if sliceLen := len(observedTimes.Timestamps); seqNum > int64(sliceLen) { + // There must have been a gap in the sequence so grow the slice + observedTimes.Timestamps = slices.Grow(observedTimes.Timestamps, int(seqNum)-sliceLen) + } + observedTimes.Timestamps[seqNum] = ts + } + observedTimes.TimestampsBySequence = nil + } + } + + // req_id->count - how many nodes reported where a new DON timestamp might be needed + observationCounts := map[string]int64{} + for _, ao := range aos { + observation := &pb.Observation{} + if err := proto.Unmarshal(ao.Observation, observation); err != nil { + p.lggr.Errorf("failed to unmarshal observation in Outcome phase") + continue + } + + for id, requestSeqNum := range observation.Requests { + var currSeqNum int64 + if times, ok := prevOutcome.ObservedDonTimes[id]; ok { + currSeqNum = int64(len(times.Timestamps)) + } + // We only count requests for the next sequence number and ignore all other ones. + if requestSeqNum == currSeqNum { + observationCounts[id]++ + } + } + } + + outcome := prevOutcome outcome.Timestamp = donTime for id, numRequests := range observationCounts { @@ -306,34 +342,78 @@ func (p *Plugin) Outcome(ctx context.Context, outctx ocr3types.OutcomeContext, _ p.store.deleteExecutionID(id) } } + return outcome +} - var outcomeBatchOverflowCount int64 - if len(outcome.ObservedDonTimes) > p.batchSize { - ids := make([]string, 0, len(outcome.ObservedDonTimes)) - for id := range outcome.ObservedDonTimes { - ids = append(ids, id) +// sequencedOutcome executed the updated outcome logic to produce a sequenced map of [pb.ObservedDonTimes.TimestampsBySequence]. +func (p *Plugin) sequencedOutcome(aos []types.AttributedObservation, prevOutcome *pb.Outcome, donTime int64) *pb.Outcome { + type reqSeq struct { + reqID string + seqNum int64 + } + // [req_id+seq_num]->count - how many nodes reported where a new DON timestamp might be needed + observationCounts := map[reqSeq]int64{} + + // At the transition point, we need to convert from the old slice format to maps + for _, observedTimes := range prevOutcome.ObservedDonTimes { + if len(observedTimes.Timestamps) > 0 { + for seqNum, ts := range observedTimes.Timestamps { + observedTimes.TimestampsBySequence[int64(seqNum)] = ts + } + observedTimes.Timestamps = nil } - slices.Sort(ids) - outcomeBatchOverflowCount = int64(len(ids) - p.batchSize) - for _, id := range ids[p.batchSize:] { - delete(outcome.ObservedDonTimes, id) + } + + for _, ao := range aos { + observation := &pb.Observation{} + if err := proto.Unmarshal(ao.Observation, observation); err != nil { + p.lggr.Errorf("failed to unmarshal observation in Outcome phase") + continue + } + + for id, requestSeqNum := range observation.Requests { + // We only count requests for future sequence numbers and ignore all other ones. + if times, ok := prevOutcome.ObservedDonTimes[id]; ok { + if requestSeqNum <= times.MaxSeqNum() { + continue + } + } + observationCounts[reqSeq{id, requestSeqNum}]++ } - p.lggr.Warnw("Trimmed outcome observed don times to batch size", - "batchSize", p.batchSize, - "removedEntries", outcomeBatchOverflowCount, - ) } - outcomeBytes, err := proto.MarshalOptions{Deterministic: true}.Marshal(outcome) - p.lggr.Infow("Outcome computed", - "observedDonTimesEntries", len(outcome.ObservedDonTimes), - "outcomeSizeBytes", len(outcomeBytes), - ) - p.metrics.donTime.Record(ctx, outcome.Timestamp) - p.metrics.donTimeEntries.Record(ctx, int64(len(outcome.ObservedDonTimes))) - p.metrics.outcomeBatchOverflow.Record(ctx, outcomeBatchOverflowCount) - p.metrics.outcomeSize.Record(ctx, int64(len(outcomeBytes))) - return outcomeBytes, err + outcome := prevOutcome + outcome.Timestamp = donTime + + for _, key := range slices.SortedFunc(maps.Keys(observationCounts), func(a, b reqSeq) int { + return cmp.Or(cmp.Compare(a.reqID, b.reqID), cmp.Compare(a.seqNum, b.seqNum)) + }) { + numRequests := observationCounts[key] + if numRequests > int64(p.config.F) { + observedDonTimes, ok := outcome.ObservedDonTimes[key.reqID] + if !ok { + observedDonTimes = &pb.ObservedDonTimes{TimestampsBySequence: make(map[int64]int64)} + } else if observedDonTimes.TimestampsBySequence == nil { + observedDonTimes.TimestampsBySequence = make(map[int64]int64) + } + observedDonTimes.TimestampsBySequence[key.seqNum] = donTime + outcome.ObservedDonTimes[key.reqID] = observedDonTimes + } + } + + // Remove expired and empty workflow executions + for id, observedTimes := range outcome.ObservedDonTimes { + if observedTimes == nil || len(observedTimes.TimestampsBySequence) == 0 { + delete(outcome.ObservedDonTimes, id) + p.store.deleteExecutionID(id) + continue + } + if ts := observedTimes.EarliestTS(); ts != nil && donTime >= *ts+p.offChainConfig.ExecutionRemovalTime.AsDuration().Milliseconds() { + delete(outcome.ObservedDonTimes, id) + p.store.deleteExecutionID(id) + } + } + return outcome } func (p *Plugin) Reports(_ context.Context, _ uint64, outcome ocr3types.Outcome) ([]ocr3types.ReportPlus[[]byte], error) { diff --git a/pkg/workflows/dontime/plugin_test.go b/pkg/workflows/dontime/plugin_test.go index 5ec9a139e2..aab3671657 100644 --- a/pkg/workflows/dontime/plugin_test.go +++ b/pkg/workflows/dontime/plugin_test.go @@ -4,6 +4,7 @@ import ( "testing" "time" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/durationpb" @@ -13,7 +14,10 @@ import ( "github.com/smartcontractkit/libocr/offchainreporting2/types" "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" + "github.com/smartcontractkit/chainlink-common/pkg/config" "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/settings" + "github.com/smartcontractkit/chainlink-common/pkg/settings/limits" "github.com/smartcontractkit/chainlink-common/pkg/workflows/dontime/pb" ) @@ -146,7 +150,7 @@ func TestPlugin_ValidateObservation(t *testing.T) { require.NoError(t, err) }) - t.Run("Invalid sequence number", func(t *testing.T) { + t.Run("Valid skipped sequence number", func(t *testing.T) { store := NewStore(DefaultRequestTimeout) plugin, err := NewPlugin(store, config, offchainCfg, lggr) require.NoError(t, err) @@ -160,24 +164,42 @@ func TestPlugin_ValidateObservation(t *testing.T) { // Add single request to queue executionID := "workflow-123" - requestCh := store.RequestDonTime(executionID, 1) + _ = store.RequestDonTime(executionID, 1) - _, err = plugin.Observation(ctx, outcomeCtx, query) + observation, err := plugin.Observation(ctx, outcomeCtx, query) require.NoError(t, err) - response := <-requestCh - require.ErrorContains(t, response.Err, "requested seqNum 1 for executionID workflow-123 is greater than the number of observed don times 0") + ao := types.AttributedObservation{ + Observation: observation, + Observer: commontypes.OracleID(1), + } + + err = plugin.ValidateObservation(ctx, outcomeCtx, query, ao) + require.NoError(t, err) }) } func TestPlugin_Outcome(t *testing.T) { + t.Run("sequenced", func(t *testing.T) { testPlugin_Outcome(t, true) }) + t.Run("unsequenced", func(t *testing.T) { testPlugin_Outcome(t, false) }) +} + +func testPlugin_Outcome(t *testing.T, sequenced bool) { lggr := logger.Test(t) store := NewStore(DefaultRequestTimeout) - config, offchainCfg := newTestPluginConfig(t), newTestPluginOffchainConfig(t) + cfg, offchainCfg := newTestPluginConfig(t), newTestPluginOffchainConfig(t) ctx := t.Context() - plugin, err := NewPlugin(store, config, offchainCfg, lggr) + plugin, err := NewPlugin(store, cfg, offchainCfg, lggr) require.NoError(t, err) + if sequenced { + plugin.setSequencedTSEnabled(limits.NewRangeLimiter( + settings.Range[config.Timestamp]{ + Lower: config.NewTimestamp(time.Now()), + Upper: config.NewTimestamp(time.Now().Add(time.Hour)), + }, + )) + } query, err := plugin.Query(ctx, ocr3types.OutcomeContext{PreviousOutcome: []byte("")}) require.NoError(t, err) @@ -233,7 +255,15 @@ func TestPlugin_Outcome(t *testing.T) { err = proto.Unmarshal(outcome, outcomeProto) require.NoError(t, err) require.Equal(t, timestamp, outcomeProto.Timestamp) - require.Equal(t, []int64{timestamp}, outcomeProto.ObservedDonTimes[executionID].Timestamps) + if observed, ok := outcomeProto.ObservedDonTimes[executionID]; assert.True(t, ok) { + if sequenced { + require.Equal(t, map[int64]int64{0: timestamp}, observed.TimestampsBySequence) + require.Empty(t, observed.Timestamps) + } else { + require.Equal(t, []int64{timestamp}, observed.Timestamps) + require.Empty(t, observed.TimestampsBySequence) + } + } } func TestPlugin_Outcome_SequenceNumberHandling(t *testing.T) { @@ -534,7 +564,7 @@ func TestPlugin_FinishedExecutions(t *testing.T) { }) t.Run("Transmit: delete removed executionIDs", func(t *testing.T) { - store.setDonTimes("workflow-123", []int64{time.Now().UnixMilli()}) + store.setDonTimes("workflow-123", map[int64]int64{0: time.Now().UnixMilli()}) r := ocr3types.ReportWithInfo[[]byte]{} r.Report, err = proto.Marshal(outcomeProto) @@ -590,7 +620,7 @@ func TestPlugin_Outcome_TrimByBatchSize(t *testing.T) { prevOutcomeBytes, err := proto.Marshal(prevOutcome) require.NoError(t, err) - t.Run("trims when all observations set batch size flag", func(t *testing.T) { + t.Run("batch size enforced", func(t *testing.T) { outcome, err := plugin.Outcome(ctx, ocr3types.OutcomeContext{PreviousOutcome: prevOutcomeBytes}, query, makeObservations(true)) require.NoError(t, err) diff --git a/pkg/workflows/dontime/store.go b/pkg/workflows/dontime/store.go index 455a0cf4b4..f5180af027 100644 --- a/pkg/workflows/dontime/store.go +++ b/pkg/workflows/dontime/store.go @@ -19,9 +19,9 @@ type Store struct { requests map[string]*Request // Maps workflow execution ID to request requestTimeout time.Duration - // donTimes holds ordered sequence timestamps generated for consecutive workflow requests - // i.e. ExecutionID --> [timestamp-0, timestamp-1 , ...] - donTimes map[string][]int64 + // donTimes holds sequence timestamps generated for workflow requests + // executionID -> sequenceNumber -> timestamp + donTimes map[string]map[int64]int64 lastObservedDonTime int64 mu sync.Mutex } @@ -30,7 +30,7 @@ func NewStore(requestTimeout time.Duration) *Store { return &Store{ requests: make(map[string]*Request), requestTimeout: requestTimeout, - donTimes: make(map[string][]int64), + donTimes: make(map[string]map[int64]int64), lastObservedDonTime: 0, mu: sync.Mutex{}, } @@ -124,30 +124,12 @@ func (s *Store) GetDonTimeForSeqNum(executionID string, seqNum int) *int64 { s.mu.Lock() defer s.mu.Unlock() if times, ok := s.donTimes[executionID]; ok { - if len(times) > seqNum { - return ×[seqNum] - } + return new(times[int64(seqNum)]) } return nil } -func (s *Store) GetDonTimes(executionID string) ([]int64, error) { - s.mu.Lock() - defer s.mu.Unlock() - - if times, ok := s.donTimes[executionID]; ok { - return times, nil - } - return []int64{}, fmt.Errorf("no don time for executionID %s", executionID) -} - -func (s *Store) setDonTimes(executionID string, donTimes []int64) { - s.mu.Lock() - defer s.mu.Unlock() - s.donTimes[executionID] = donTimes -} - -func (s *Store) replaceDonTimes(donTimes map[string][]int64) { +func (s *Store) replaceDonTimes(donTimes map[string]map[int64]int64) { s.mu.Lock() defer s.mu.Unlock() diff --git a/pkg/workflows/dontime/store_test.go b/pkg/workflows/dontime/store_test.go index b5ff631dfd..a471586dba 100644 --- a/pkg/workflows/dontime/store_test.go +++ b/pkg/workflows/dontime/store_test.go @@ -1,6 +1,7 @@ package dontime import ( + "fmt" "testing" "time" @@ -24,3 +25,19 @@ func TestStore_RequestExpiresWithoutPlugin(t *testing.T) { require.Nil(t, store.GetRequest(executionID)) } + +func (s *Store) GetDonTimes(executionID string) (map[int64]int64, error) { + s.mu.Lock() + defer s.mu.Unlock() + + if times, ok := s.donTimes[executionID]; ok { + return times, nil + } + return map[int64]int64{}, fmt.Errorf("no don time for executionID %s", executionID) +} + +func (s *Store) setDonTimes(executionID string, donTimes map[int64]int64) { + s.mu.Lock() + defer s.mu.Unlock() + s.donTimes[executionID] = donTimes +} diff --git a/pkg/workflows/dontime/transmitter.go b/pkg/workflows/dontime/transmitter.go index 8aa2ceb943..9622e76f2b 100644 --- a/pkg/workflows/dontime/transmitter.go +++ b/pkg/workflows/dontime/transmitter.go @@ -5,10 +5,11 @@ import ( "google.golang.org/protobuf/proto" - "github.com/smartcontractkit/chainlink-common/pkg/logger" - "github.com/smartcontractkit/chainlink-common/pkg/workflows/dontime/pb" "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" "github.com/smartcontractkit/libocr/offchainreporting2plus/types" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/workflows/dontime/pb" ) var _ ocr3types.ContractTransmitter[[]byte] = (*Transmitter)(nil) @@ -33,16 +34,29 @@ func (t *Transmitter) Transmit(_ context.Context, _ types.ConfigDigest, _ uint64 return err } - currentDonTimes := make(map[string][]int64, len(outcome.ObservedDonTimes)) + var totalCount int + entriesCount := make(map[string]map[int64]int64, len(outcome.ObservedDonTimes)) for id, observedDonTimes := range outcome.ObservedDonTimes { - currentDonTimes[id] = observedDonTimes.Timestamps + if len(observedDonTimes.Timestamps) > 0 { + m := make(map[int64]int64) + for i, donTime := range observedDonTimes.Timestamps { + if donTime == 0 { // feature flag was disabled, and we had a gap in the sequence + continue + } + m[int64(i)] = donTime + } + entriesCount[id] = m + } else { + entriesCount[id] = observedDonTimes.TimestampsBySequence + } + totalCount += len(entriesCount[id]) } - t.store.replaceDonTimes(currentDonTimes) + t.store.replaceDonTimes(entriesCount) t.store.setLastObservedDonTime(outcome.Timestamp) - t.lggr.Infow("Transmitting timestamps", "lastObservedDonTime", outcome.Timestamp) + t.lggr.Infow("Transmitting timestamps", "lastObservedDonTime", outcome.Timestamp, "donTimeEntries", len(entriesCount), "donTimeTotal", totalCount) - for executionID, donTimes := range outcome.ObservedDonTimes { + for executionID, donTimes := range entriesCount { request := t.store.GetRequest(executionID) if request == nil { continue @@ -50,8 +64,7 @@ func (t *Transmitter) Transmit(_ context.Context, _ types.ConfigDigest, _ uint64 // Nodes behind on multiple requests may wait one OCR round per request. // Caching future times locally could be added as an optimization. - if len(donTimes.Timestamps) > request.SeqNum { - donTime := donTimes.Timestamps[request.SeqNum] + if donTime, ok := donTimes[int64(request.SeqNum)]; ok { t.store.RemoveRequest(executionID) // Make space for next request before delivering request.SendResponse(Response{ WorkflowExecutionID: executionID, diff --git a/pkg/workflows/dontime/transmitter_test.go b/pkg/workflows/dontime/transmitter_test.go index 18f9023498..31c12eb39e 100644 --- a/pkg/workflows/dontime/transmitter_test.go +++ b/pkg/workflows/dontime/transmitter_test.go @@ -7,10 +7,11 @@ import ( "github.com/stretchr/testify/require" "google.golang.org/protobuf/proto" - "github.com/smartcontractkit/chainlink-common/pkg/logger" - "github.com/smartcontractkit/chainlink-common/pkg/workflows/dontime/pb" "github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types" "github.com/smartcontractkit/libocr/offchainreporting2plus/types" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/workflows/dontime/pb" ) func TestTransmitter_TransmitDonTimeRequest(t *testing.T) { @@ -20,34 +21,45 @@ func TestTransmitter_TransmitDonTimeRequest(t *testing.T) { transmitter := NewTransmitter(lggr, store, "") - // Create request for second donTime in sequence - executionID := "workflow-123" - timeRequest := store.RequestDonTime(executionID, 1) - timestamp := time.Now().UnixMilli() - outcome := &pb.Outcome{ - Timestamp: timestamp, - ObservedDonTimes: map[string]*pb.ObservedDonTimes{ - executionID: {Timestamps: []int64{timestamp - int64(time.Second), timestamp}}, - }, - } - r := ocr3types.ReportWithInfo[[]byte]{} - var err error - r.Report, err = proto.Marshal(outcome) - require.NoError(t, err) - err = transmitter.Transmit(ctx, types.ConfigDigest{}, 0, r, []types.AttributedOnchainSignature{}) - require.NoError(t, err) - - select { - case donTimeResp := <-timeRequest: - require.Equal(t, timestamp, donTimeResp.Timestamp) - require.Equal(t, executionID, donTimeResp.WorkflowExecutionID) - require.Equal(t, 1, donTimeResp.SeqNum) - require.NoError(t, donTimeResp.Err) - case <-ctx.Done(): - t.Fatal("failed to retrieve donTime from request channel") - } + for _, tc := range []struct { + name string + observed *pb.ObservedDonTimes + }{ + {"unsequenced", &pb.ObservedDonTimes{Timestamps: []int64{timestamp - int64(time.Second), timestamp}}}, + {"sequenced", &pb.ObservedDonTimes{TimestampsBySequence: map[int64]int64{0: timestamp - int64(time.Second), 1: timestamp}}}, + {"both", &pb.ObservedDonTimes{Timestamps: []int64{timestamp - int64(time.Second), timestamp}, + TimestampsBySequence: map[int64]int64{0: timestamp - int64(time.Second), 1: timestamp}}}, + } { + t.Run(tc.name, func(t *testing.T) { + // Create request for second donTime in sequence + executionID := "workflow-123" + timeRequest := store.RequestDonTime(executionID, 1) + + outcome := &pb.Outcome{ + Timestamp: timestamp, + ObservedDonTimes: map[string]*pb.ObservedDonTimes{executionID: tc.observed}, + } - require.Empty(t, store.GetRequest(executionID)) + r := ocr3types.ReportWithInfo[[]byte]{} + var err error + r.Report, err = proto.Marshal(outcome) + require.NoError(t, err) + err = transmitter.Transmit(ctx, types.ConfigDigest{}, 0, r, []types.AttributedOnchainSignature{}) + require.NoError(t, err) + + select { + case donTimeResp := <-timeRequest: + require.Equal(t, timestamp, donTimeResp.Timestamp) + require.Equal(t, executionID, donTimeResp.WorkflowExecutionID) + require.Equal(t, 1, donTimeResp.SeqNum) + require.NoError(t, donTimeResp.Err) + case <-ctx.Done(): + t.Fatal("failed to retrieve donTime from request channel") + } + + require.Empty(t, store.GetRequest(executionID)) + }) + } }