omg im so dumb

mod/async/pkg/publisher/publisher.go

@@ -84,7 +84,7 @@ func (p *Publisher[T]) start(ctx context.Context) {
 func (p *Publisher[T]) Publish(msg types.BaseEvent) error {
 	typedMsg, err := ensureType[T](msg)
 	if err != nil {
-		return errIncompatibleAssignee(msg, *new(T))
+		return err
 	}
 	ctx := msg.Context()
 	select {

mod/async/pkg/types/event.go

@@ -41,18 +41,18 @@ type Event[DataT any] interface {
 	BaseEvent
 	Data() DataT
 	Error() error
-	Is(messageType EventID) bool
+	Is(id EventID) bool
 }
 
 // NewEvent creates a new Event with the given context and beacon event.
 func NewEvent[
 	DataT any,
 ](
-	ctx context.Context, messageType EventID, data DataT, errs ...error,
+	ctx context.Context, id EventID, data DataT, errs ...error,
 ) Event[DataT] {
 	return &event[DataT]{
 		ctx:  ctx,
-		id:   messageType,
+		id:   id,
 		data: data,
 		err:  errors.Join(errs...),
 	}

mod/async/pkg/types/subscription.go

@@ -22,7 +22,6 @@ package types
 
 import (
 	"context"
-	"fmt"
 )
 
 // Subscription is a channel that receives events.
@@ -63,10 +62,8 @@ func (s Subscription[T]) Listen(
 // Await will block until an event is received from the subscription or the
 // context is canceled.
 func (s Subscription[T]) Await(ctx context.Context) (T, error) {
-	fmt.Println("AWAITING EVENT")
 	select {
 	case event := <-s:
-		fmt.Println("GOT EVENT")
 		return event, nil
 	case <-ctx.Done():
 		return *new(T), ctx.Err()

mod/beacon/blockchain/service.go

@@ -22,7 +22,6 @@ package blockchain
 
 import (
 	"context"
-	"fmt"
 	"sync"
 
 	async "github.com/berachain/beacon-kit/mod/async/pkg/types"
@@ -227,7 +226,6 @@ func (s *Service[
 	if genesisErr != nil {
 		s.logger.Error("Failed to process genesis data", "error", genesisErr)
 	}
-	fmt.Println("GEN DATA PROCESSED")
 
 	// Emit the event containing the validator updates.
 	if err := s.dispatcher.PublishEvent(
@@ -244,7 +242,6 @@ func (s *Service[
 		)
 		panic(err)
 	}
-	fmt.Println("EMITTED EVENT")
 }
 
 func (s *Service[

mod/node-core/pkg/components/service_registry.go

@@ -32,6 +32,7 @@ type ServiceRegistryInput[
 	LoggerT log.AdvancedLogger[any, LoggerT],
 ] struct {
 	depinject.In
+	Middleware        *ABCIMiddleware
 	BlockStoreService *BlockStoreService
 	ChainService      *ChainService
 	DAService         *DAService
@@ -54,6 +55,7 @@ func ProvideServiceRegistry[
 ) *service.Registry {
 	return service.NewRegistry(
 		service.WithLogger(in.Logger),
+		service.WithService(in.Middleware),
 		service.WithService(in.Dispatcher),
 		service.WithService(in.ValidatorService),
 		service.WithService(in.BlockStoreService),

mod/runtime/pkg/middleware/abci.go

@@ -22,7 +22,6 @@ package middleware
 
 import (
 	"context"
-	"fmt"
 	"time"
 
 	async "github.com/berachain/beacon-kit/mod/async/pkg/types"
@@ -48,9 +47,11 @@ func (h *ABCIMiddleware[
 	bz []byte,
 ) (transition.ValidatorUpdates, error) {
 	var (
-		err      error
-		gdpEvent async.Event[transition.ValidatorUpdates]
+		err              error
+		gdpEvent         async.Event[transition.ValidatorUpdates]
+		awaitCtx, cancel = context.WithTimeout(ctx, AwaitTimeout)
 	)
+	defer cancel()
 	// in theory this channel should already be empty, but we clear it anyways
 	h.subGenDataProcessed.Clear()
 
@@ -66,11 +67,11 @@ func (h *ABCIMiddleware[
 		return nil, err
 	}
 
-	gdpEvent, err = h.subGenDataProcessed.Await(ctx)
+	gdpEvent, err = h.subGenDataProcessed.Await(awaitCtx)
 	if err != nil {
 		return nil, err
 	}
-	fmt.Println("GOT IT BACK")
+
 	return gdpEvent.Data(), gdpEvent.Error()
 }
 
@@ -86,11 +87,13 @@ func (h *ABCIMiddleware[
 	slotData SlotDataT,
 ) ([]byte, []byte, error) {
 	var (
-		err          error
-		builtBBEvent async.Event[BeaconBlockT]
-		builtSCEvent async.Event[BlobSidecarsT]
-		startTime    = time.Now()
+		err              error
+		builtBBEvent     async.Event[BeaconBlockT]
+		builtSCEvent     async.Event[BlobSidecarsT]
+		startTime        = time.Now()
+		awaitCtx, cancel = context.WithTimeout(ctx, AwaitTimeout)
 	)
+	defer cancel()
 	defer h.metrics.measurePrepareProposalDuration(startTime)
 	// in theory these subs should already be empty, but we clear them anyways
 	h.subBuiltBeaconBlock.Clear()
@@ -105,7 +108,7 @@ func (h *ABCIMiddleware[
 	}
 
 	// wait for built beacon block
-	builtBBEvent, err = h.subBuiltBeaconBlock.Await(ctx)
+	builtBBEvent, err = h.subBuiltBeaconBlock.Await(awaitCtx)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -114,7 +117,7 @@ func (h *ABCIMiddleware[
 	}
 
 	// wait for built sidecars
-	builtSCEvent, err = h.subBuiltSidecars.Await(ctx)
+	builtSCEvent, err = h.subBuiltSidecars.Await(awaitCtx)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -165,12 +168,14 @@ func (h *ABCIMiddleware[
 	req proto.Message,
 ) (proto.Message, error) {
 	var (
-		err       error
-		startTime = time.Now()
-		blk       BeaconBlockT
-		sidecars  BlobSidecarsT
+		err              error
+		startTime        = time.Now()
+		blk              BeaconBlockT
+		sidecars         BlobSidecarsT
+		awaitCtx, cancel = context.WithTimeout(ctx, AwaitTimeout)
 	)
-	// in theory these subs should already be empty, but we clear them anyways
+	defer cancel()
+	// in theory these subs should already be empty, probably redundant
 	h.subBBVerified.Clear()
 	h.subSCVerified.Clear()
 	abciReq, ok := req.(*cmtabci.ProcessProposalRequest)
@@ -185,7 +190,6 @@ func (h *ABCIMiddleware[
 		return h.createProcessProposalResponse(errors.WrapNonFatal(err))
 	}
 
-	// TODO: implement service
 	// notify that the beacon block has been received.
 	if err = h.dispatcher.PublishEvent(
 		async.NewEvent(ctx, events.BeaconBlockReceived, blk),
@@ -206,11 +210,11 @@ func (h *ABCIMiddleware[
 	}
 
 	// err if the built beacon block or sidecars failed verification.
-	_, err = h.subBBVerified.Await(ctx)
+	_, err = h.subBBVerified.Await(awaitCtx)
 	if err != nil {
 		return h.createProcessProposalResponse(err)
 	}
-	_, err = h.subSCVerified.Await(ctx)
+	_, err = h.subSCVerified.Await(awaitCtx)
 	if err != nil {
 		return h.createProcessProposalResponse(err)
 	}
@@ -245,7 +249,9 @@ func (h *ABCIMiddleware[
 		blk                  BeaconBlockT
 		blobs                BlobSidecarsT
 		finalValUpdatesEvent async.Event[transition.ValidatorUpdates]
+		awaitCtx, cancel     = context.WithTimeout(ctx, AwaitTimeout)
 	)
+	defer cancel()
 	// in theory this sub should already be empty, but we clear them anyways
 	h.subFinalValidatorUpdates.Clear()
 	abciReq, ok := req.(*cmtabci.FinalizeBlockRequest)
@@ -281,7 +287,7 @@ func (h *ABCIMiddleware[
 	}
 
 	// wait for the final validator updates.
-	finalValUpdatesEvent, err = h.subFinalValidatorUpdates.Await(ctx)
+	finalValUpdatesEvent, err = h.subFinalValidatorUpdates.Await(awaitCtx)
 	if err != nil {
 		return nil, err
 	}

mod/runtime/pkg/middleware/constants.go

@@ -20,11 +20,15 @@
 
 package middleware
 
+import "time"
+
 const (
 	// BeaconBlockTxIndex represents the index of the beacon block transaction.
 	// It is the first transaction in the tx list.
 	BeaconBlockTxIndex uint = iota
 	// BlobSidecarsTxIndex represents the index of the blob sidecar transaction.
 	// It follows the beacon block transaction in the tx list.
 	BlobSidecarsTxIndex
+	// AwaitTimeout is the timeout for awaiting events.
+	AwaitTimeout = 2 * time.Second
 )

mod/runtime/pkg/middleware/middleware.go

@@ -21,6 +21,7 @@
 package middleware
 
 import (
+	"context"
 	"encoding/json"
 
 	"github.com/berachain/beacon-kit/mod/async/pkg/types"
@@ -133,34 +134,34 @@ func NewABCIMiddleware[
 func (am *ABCIMiddleware[
 	AvailabilityStoreT, BeaconBlockT, BeaconBlockBundleT, BlobSidecarsT,
 	DepositT, ExecutionPayloadT, GenesisT, SlotDataT,
-]) Start() error {
-	// subGenDat
-	if err := am.dispatcher.Subscribe(
+]) Start(_ context.Context) error {
+	var err error
+	if err = am.dispatcher.Subscribe(
 		events.GenesisDataProcessed, am.subGenDataProcessed,
 	); err != nil {
 		return err
 	}
-	if err := am.dispatcher.Subscribe(
+	if err = am.dispatcher.Subscribe(
 		events.BuiltBeaconBlock, am.subBuiltBeaconBlock,
 	); err != nil {
 		return err
 	}
-	if err := am.dispatcher.Subscribe(
+	if err = am.dispatcher.Subscribe(
 		events.BuiltSidecars, am.subBuiltSidecars,
 	); err != nil {
 		return err
 	}
-	if err := am.dispatcher.Subscribe(
+	if err = am.dispatcher.Subscribe(
 		events.BeaconBlockVerified, am.subBBVerified,
 	); err != nil {
 		return err
 	}
-	if err := am.dispatcher.Subscribe(
+	if err = am.dispatcher.Subscribe(
 		events.SidecarsVerified, am.subSCVerified,
 	); err != nil {
 		return err
 	}
-	if err := am.dispatcher.Subscribe(
+	if err = am.dispatcher.Subscribe(
 		events.FinalValidatorUpdatesProcessed, am.subFinalValidatorUpdates,
 	); err != nil {
 		return err