DonTime module hooks

pkg/workflows/dontime/factory.go

@@ -33,7 +33,7 @@ type Factory struct {
 func NewFactory(s *Store, lggr logger.Logger) (*Factory, error) {
 	return &Factory{
 		store: s,
-		lggr:  logger.Named(lggr, "OCR3WorkflowLibFactory"),
+		lggr:  logger.Named(lggr, "OCR3DonTimeFactory"),
 	}, nil
 }
 
@@ -84,13 +84,13 @@ func (o *Factory) NewReportingPlugin(_ context.Context, config ocr3types.Reporti
 }
 
 func (o *Factory) Start(ctx context.Context) error {
-	return o.StartOnce("OCR3WorkflowLibPlugin", func() error {
+	return o.StartOnce("OCR3DonTimePlugin", func() error {
 		return nil
 	})
 }
 
 func (o *Factory) Close() error {
-	return o.StopOnce("OCR3WorkflowLibPlugin", func() error {
+	return o.StopOnce("OCR3DonTimePlugin", func() error {
 		return nil
 	})
 }

pkg/workflows/dontime/plugin_test.go

@@ -260,7 +260,7 @@ func TestPlugin_FinishedExecutions(t *testing.T) {
 	config := newTestPluginConfig(t)
 	ctx := t.Context()
 
-	transmitter := NewTransmitter(lggr, store, defaultBatchSize)
+	transmitter := NewTransmitter(lggr, store)
 	plugin, err := NewPlugin(store, config, lggr)
 	require.NoError(t, err)
 

pkg/workflows/dontime/provider.go

@@ -0,0 +1,45 @@
+package dontime
+
+import (
+	"context"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/services"
+	"github.com/smartcontractkit/libocr/offchainreporting2plus/ocr3types"
+)
+
+// Provider wraps an existing OCR3 plugin provider (from the relayer)
+// and overrides the plugin factory and contract transmitter for DonTime.
+type Provider struct {
+	Factory     ocr3types.ReportingPluginFactory[struct{}]
+	Transmitter ocr3types.ContractTransmitter[struct{}]
+}
+
+var _ services.Service = &Provider{}
+
+func (p *Provider) Start(_ context.Context) error {
+	return nil
+}
+
+func (p *Provider) Close() error {
+	return nil
+}
+
+func (p *Provider) Name() string {
+	return "DonTimeOCR3Provider"
+}
+
+func (p *Provider) HealthReport() map[string]error {
+	return map[string]error{p.Name(): nil}
+}
+
+func (p *Provider) Ready() error {
+	return nil
+}
+
+func (p *Provider) ReportingPluginFactory() ocr3types.ReportingPluginFactory[struct{}] {
+	return p.Factory
+}
+
+func (p *Provider) ContractTransmitter() ocr3types.ContractTransmitter[struct{}] {
+	return p.Transmitter
+}

pkg/workflows/dontime/transmitter.go

@@ -16,13 +16,12 @@ var _ ocr3types.ContractTransmitter[struct{}] = (*Transmitter)(nil)
 // When called it will transmit DonTime requests back to the caller
 // and handle deletion of finished executionIDs.
 type Transmitter struct {
-	lggr      logger.Logger
-	store     *Store
-	batchSize int
+	lggr  logger.Logger
+	store *Store
 }
 
-func NewTransmitter(lggr logger.Logger, store *Store, batchSize int) *Transmitter {
-	return &Transmitter{lggr: lggr, store: store, batchSize: batchSize}
+func NewTransmitter(lggr logger.Logger, store *Store) *Transmitter {
+	return &Transmitter{lggr: lggr, store: store}
 }
 
 func (t *Transmitter) Transmit(_ context.Context, _ types.ConfigDigest, _ uint64, r ocr3types.ReportWithInfo[struct{}], _ []types.AttributedOnchainSignature) error {

pkg/workflows/dontime/transmitter_test.go

@@ -18,7 +18,7 @@ func TestTransmitter_TransmitDonTimeRequest(t *testing.T) {
 	store := NewStore(DefaultRequestTimeout)
 	ctx := t.Context()
 
-	transmitter := NewTransmitter(lggr, store, defaultBatchSize)
+	transmitter := NewTransmitter(lggr, store)
 
 	// Create request for second donTime in sequence
 	executionID := "workflow-123"

pkg/workflows/wasm/host/execution.go

@@ -2,8 +2,10 @@ package host
 
 import (
 	"context"
+	"encoding/binary"
 	"fmt"
 	"sync"
+	"time"
 
 	"github.com/bytecodealliance/wasmtime-go/v28"
 	sdkpb "github.com/smartcontractkit/chainlink-common/pkg/workflows/sdk/v2/pb"
@@ -18,6 +20,8 @@ type execution[T any] struct {
 	lock                 sync.RWMutex
 	module               *module
 	executor             ExecutionHelper
+	timeFetcher          *timeFetcher
+	baseTime             *time.Time
 	hasRun               bool
 	mode                 sdkpb.Mode
 	donSeed              int64
@@ -161,3 +165,111 @@ func (e *execution[T]) switchModes(_ *wasmtime.Caller, mode int32) {
 	e.hasRun = true
 	e.mode = sdkpb.Mode(mode)
 }
+
+func (e *execution[T]) clockTimeGet(caller *wasmtime.Caller, id int32, precision int64, resultTimestamp int32) int32 {
+	donTime, err := e.timeFetcher.GetTime(e.mode)
+	if err != nil {
+		return ErrnoInval
+	}
+
+	if e.baseTime == nil {
+		// baseTime must be before the first poll or Go panics
+		t := donTime.Add(-time.Nanosecond)
+		e.baseTime = &t
+	}
+
+	var val int64
+	switch id {
+	case clockIDMonotonic:
+		val = donTime.Sub(*e.baseTime).Nanoseconds()
+	case clockIDRealtime:
+		val = donTime.UnixNano()
+	default:
+		return ErrnoInval
+	}
+
+	uint64Size := int32(8)
+	trg := make([]byte, uint64Size)
+	binary.LittleEndian.PutUint64(trg, uint64(val))
+	wasmWrite(caller, trg, resultTimestamp, uint64Size)
+	return ErrnoSuccess
+}
+
+// Loosely based off the implementation here:
+// https://github.com/tetratelabs/wazero/blob/main/imports/wasi_snapshot_preview1/poll.go#L52
+// For an overview of the spec, including the datatypes being referred to, see:
+// https://github.com/WebAssembly/WASI/blob/snapshot-01/phases/snapshot/docs.md
+// This implementation only responds to clock events, not to file descriptor notifications.
+// It sleeps based on the largest timeout
+func (e *execution[T]) pollOneoff(caller *wasmtime.Caller, subscriptionptr int32, eventsptr int32, nsubscriptions int32, resultNevents int32) int32 {
+	if nsubscriptions == 0 {
+		return ErrnoInval
+	}
+
+	subs, err := wasmRead(caller, subscriptionptr, nsubscriptions*subscriptionLen)
+	if err != nil {
+		return ErrnoFault
+	}
+
+	events := make([]byte, nsubscriptions*eventsLen)
+	timeout := time.Duration(0)
+
+	for i := int32(0); i < nsubscriptions; i++ {
+		inOffset := i * subscriptionLen
+		userData := subs[inOffset : inOffset+8]
+		eventType := subs[inOffset+8]
+		argBuf := subs[inOffset+8+8:]
+
+		slot, err := getSlot(events, i)
+		if err != nil {
+			return ErrnoFault
+		}
+
+		switch eventType {
+		case eventTypeClock:
+			newTimeout := binary.LittleEndian.Uint64(argBuf[8:16])
+			flag := binary.LittleEndian.Uint16(argBuf[24:32])
+
+			var errno Errno
+			switch flag {
+			case 0: // relative
+				errno = ErrnoSuccess
+				if timeout < time.Duration(newTimeout) {
+					timeout = time.Duration(newTimeout)
+				}
+			default:
+				errno = ErrnoNotsup
+			}
+			writeEvent(slot, userData, errno, eventTypeClock)
+
+		case eventTypeFDRead, eventTypeFDWrite:
+			writeEvent(slot, userData, ErrnoBadf, int(eventType))
+
+		default:
+			writeEvent(slot, userData, ErrnoInval, int(eventType))
+		}
+	}
+
+	if timeout > 0 {
+		select {
+		case <-time.After(timeout):
+		case <-e.ctx.Done():
+			// If context was cancelled, there will be a trap from the engine
+			// which will halt execution, therefore the return value isn't read
+			return 0
+		}
+	}
+
+	uint32Size := int32(4)
+	rne := make([]byte, uint32Size)
+	binary.LittleEndian.PutUint32(rne, uint32(nsubscriptions))
+
+	if wasmWrite(caller, rne, resultNevents, uint32Size) == -1 {
+		return ErrnoFault
+	}
+	if wasmWrite(caller, events, eventsptr, nsubscriptions*eventsLen) == -1 {
+		return ErrnoFault
+	}
+
+	return ErrnoSuccess
+}

pkg/workflows/wasm/host/module.go

@@ -100,7 +100,7 @@ type ExecutionHelper interface {
 
 	GetNodeTime() time.Time
 
-	GetDONTime() time.Time
+	GetDONTime(ctx context.Context) (time.Time, error)
 
 	EmitUserLog(log string) error
 }

pkg/workflows/wasm/host/module_test.go

@@ -3,14 +3,13 @@ package host
 import (
 	"context"
 	"encoding/binary"
-	"sync"
-	"testing"
-
 	"github.com/bytecodealliance/wasmtime-go/v28"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
+	"sync"
+	"testing"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/custmsg"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"