runtime: subscription engine refactor [1]

This change is the first in a series of refactors to the subscription
engine. This work is in preparation for adding Actor Subscriptions as
detailed [here](dapr/proposals#78). In order to
implement this new subscription type, the engine needs a hierarchical
reordering of the internal data structures in flow to make adding the
new implementation sane and maintainable.

This first change introduces a new `Postman` interface that is used to
send messages to subscription sinks- HTTP, gRPC or streaming channels.
Actors would be another sink added in future. Shared types have been
moved to a new `pkg/runtime/subscription/todo` package.

Appreciate that there will be further changes made to this code path.

Signed-off-by: joshvanl <[email protected]>

Author: JoshVanL

pkg/runtime/processor/subscriber/subscriber.go

@@ -32,6 +32,10 @@ import (
 	"github.com/dapr/dapr/pkg/runtime/compstore"
 	rtpubsub "github.com/dapr/dapr/pkg/runtime/pubsub"
 	"github.com/dapr/dapr/pkg/runtime/subscription"
+	"github.com/dapr/dapr/pkg/runtime/subscription/postman"
+	postmangrpc "github.com/dapr/dapr/pkg/runtime/subscription/postman/grpc"
+	"github.com/dapr/dapr/pkg/runtime/subscription/postman/http"
+	"github.com/dapr/dapr/pkg/runtime/subscription/postman/streaming"
 	"github.com/dapr/kit/logger"
 )
 
@@ -492,24 +496,43 @@ func (s *Subscriber) initProgramaticSubscriptions(ctx context.Context) error {
 }
 
 func (s *Subscriber) startSubscription(pubsub *rtpubsub.PubsubItem, comp *compstore.NamedSubscription, isStreamer bool) (*subscription.Subscription, error) {
+	// TODO: @joshvanl
+	var postman postman.Interface
 	var streamer rtpubsub.AdapterStreamer
 	if isStreamer {
 		streamer = s.adapterStreamer
+		postman = streaming.New(streaming.Options{
+			Tracing: s.tracingSpec,
+			Channel: s.adapterStreamer,
+		})
+	} else {
+		if s.isHTTP {
+			postman = http.New(http.Options{
+				Channels: s.channels,
+				Tracing:  s.tracingSpec,
+				Adapter:  s.adapter,
+			})
+		} else {
+			postman = postmangrpc.New(postmangrpc.Options{
+				Channel: s.grpc,
+				Tracing: s.tracingSpec,
+				Adapter: s.adapter,
+			})
+		}
 	}
 	return subscription.New(subscription.Options{
 		AppID:           s.appID,
 		Namespace:       s.namespace,
 		PubSubName:      comp.PubsubName,
 		Topic:           comp.Topic,
-		IsHTTP:          s.isHTTP,
 		PubSub:          pubsub,
 		Resiliency:      s.resiliency,
 		TraceSpec:       s.tracingSpec,
 		Route:           comp.Subscription,
-		Channels:        s.channels,
 		GRPC:            s.grpc,
 		Adapter:         s.adapter,
 		AdapterStreamer: streamer,
 		ConnectionID:    comp.ConnectionID,
+		Postman:         postman,
 	})
 }

pkg/runtime/pubsub/adapter.go

@@ -45,7 +45,7 @@ type Adapter interface {
 
 type AdapterStreamer interface {
 	Subscribe(rtv1pb.Dapr_SubscribeTopicEventsAlpha1Server, *rtv1pb.SubscribeTopicEventsRequestInitialAlpha1, ConnectionID) error
-	Publish(context.Context, *SubscribedMessage) error
+	Publish(context.Context, *SubscribedMessage) (*rtv1pb.SubscribeTopicEventsRequestProcessedAlpha1, error)
 	StreamerKey(pubsub, topic string) string
 	Close(key string, connectionID ConnectionID)
 }

pkg/runtime/pubsub/streamer/streamer.go

@@ -146,27 +146,28 @@ func (s *streamer) recvLoop(
 	}
 }
 
-func (s *streamer) Publish(ctx context.Context, msg *rtpubsub.SubscribedMessage) error {
+// TODO: @joshvanl: move diagnostics.
+func (s *streamer) Publish(ctx context.Context, msg *rtpubsub.SubscribedMessage) (*rtv1pb.SubscribeTopicEventsRequestProcessedAlpha1, error) {
 	s.lock.RLock()
 	key := s.StreamerKey(msg.PubSub, msg.Topic)
 	connection, ok := s.subscribers[key][msg.SubscriberID]
 	s.lock.RUnlock()
 	if !ok {
-		return fmt.Errorf("no streamer subscribed to pubsub %q topic %q", msg.PubSub, msg.Topic)
+		return nil, fmt.Errorf("no streamer subscribed to pubsub %q topic %q", msg.PubSub, msg.Topic)
 	}
 
 	if connection.closed.Load() {
-		return errors.New("connection is closed")
+		return nil, errors.New("connection is closed")
 	}
 
 	envelope, span, err := rtpubsub.GRPCEnvelopeFromSubscriptionMessage(ctx, msg, log, s.tracingSpec)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	ch, cleanup := connection.registerPublishResponse(envelope.GetId())
 	if ch == nil {
-		return fmt.Errorf("no client stream expecting publish response for id %s ConnectionID%d", envelope.GetId(), connection.connectionID)
+		return nil, fmt.Errorf("no client stream expecting publish response for id %s ConnectionID%d", envelope.GetId(), connection.connectionID)
 	}
 	defer cleanup()
 
@@ -191,35 +192,16 @@ func (s *streamer) Publish(ctx context.Context, msg *rtpubsub.SubscribedMessage)
 		err = fmt.Errorf("error returned from app while processing pub/sub event %v: %w", msg.CloudEvent[contribpubsub.IDField], rterrors.NewRetriable(err))
 		log.Debug(err)
 		diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.PubSub, strings.ToLower(string(contribpubsub.Retry)), "", msg.Topic, elapsed)
-		return err
+		return nil, err
 	}
 
-	var resp *rtv1pb.SubscribeTopicEventsRequestProcessedAlpha1
 	select {
 	case <-ctx.Done():
-		return ctx.Err()
+		return nil, ctx.Err()
 	case <-connection.closeCh:
-	case resp = <-ch:
-	}
-
-	switch resp.GetStatus().GetStatus() {
-	case rtv1pb.TopicEventResponse_SUCCESS: //nolint:nosnakecase
-		// on uninitialized status, this is the case it defaults to as an uninitialized status defaults to 0 which is
-		// success from protobuf definition
-		diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.PubSub, strings.ToLower(string(contribpubsub.Success)), "", msg.Topic, elapsed)
-		return nil
-	case rtv1pb.TopicEventResponse_RETRY: //nolint:nosnakecase
-		diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.PubSub, strings.ToLower(string(contribpubsub.Retry)), "", msg.Topic, elapsed)
-		// TODO: add retry error info
-		return fmt.Errorf("RETRY status returned from app while processing pub/sub event %v: %w", msg.CloudEvent[contribpubsub.IDField], rterrors.NewRetriable(nil))
-	case rtv1pb.TopicEventResponse_DROP: //nolint:nosnakecase
-		log.Warnf("DROP status returned from app while processing pub/sub event %v", msg.CloudEvent[contribpubsub.IDField])
-		diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.PubSub, strings.ToLower(string(contribpubsub.Drop)), "", msg.Topic, elapsed)
-		return rtpubsub.ErrMessageDropped
-	default:
-		// Consider unknown status field as error and retry
-		diag.DefaultComponentMonitoring.PubsubIngressEvent(ctx, msg.PubSub, strings.ToLower(string(contribpubsub.Retry)), "", msg.Topic, elapsed)
-		return fmt.Errorf("unknown status returned from app while processing pub/sub event %v, status: %v, err: %w", msg.CloudEvent[contribpubsub.IDField], resp.GetStatus(), rterrors.NewRetriable(nil))
+		return nil, errors.New("stream closed")
+	case resp := <-ch:
+		return resp, nil
 	}
 }
 

pkg/runtime/subscription/bulkresiliency.go

@@ -19,60 +19,59 @@ import (
 
 	contribpubsub "github.com/dapr/components-contrib/pubsub"
 	"github.com/dapr/dapr/pkg/resiliency"
+	"github.com/dapr/dapr/pkg/runtime/subscription/postman"
+	"github.com/dapr/dapr/pkg/runtime/subscription/todo"
 	"github.com/dapr/dapr/utils"
 )
 
-type bulkSubscribeResiliencyRes struct {
-	entries  []contribpubsub.BulkSubscribeResponseEntry
-	envelope map[string]interface{}
-}
-
 // applyBulkSubscribeResiliency applies resiliency support to bulk subscribe. It tries to filter
 // out the messages that have been successfully processed and only retries the ones that have failed
-func (s *Subscription) applyBulkSubscribeResiliency(ctx context.Context, bulkSubCallData *bulkSubscribeCallData,
-	psm bulkSubscribedMessage, deadLetterTopic string, path string, policyDef *resiliency.PolicyDefinition,
+func (s *Subscription) applyBulkSubscribeResiliency(ctx context.Context, bulkSubCallData *todo.BulkSubscribeCallData,
+	psm todo.BulkSubscribedMessage, deadLetterTopic string, path string, policyDef *resiliency.PolicyDefinition,
 	rawPayload bool, envelope map[string]interface{},
 ) (*[]contribpubsub.BulkSubscribeResponseEntry, error) {
 	bscData := *bulkSubCallData
 	policyRunner := resiliency.NewRunnerWithOptions(
-		ctx, policyDef, resiliency.RunnerOpts[*bulkSubscribeResiliencyRes]{
-			Accumulator: func(bsrr *bulkSubscribeResiliencyRes) {
-				for _, v := range bsrr.entries {
+		ctx, policyDef, resiliency.RunnerOpts[*todo.BulkSubscribeResiliencyRes]{
+			Accumulator: func(bsrr *todo.BulkSubscribeResiliencyRes) {
+				for _, v := range bsrr.Entries {
 					// add to main bulkResponses
-					if index, ok := (*bscData.entryIdIndexMap)[v.EntryId]; ok {
-						(*bscData.bulkResponses)[index].EntryId = v.EntryId
-						(*bscData.bulkResponses)[index].Error = v.Error
+					if index, ok := (*bscData.EntryIdIndexMap)[v.EntryId]; ok {
+						(*bscData.BulkResponses)[index].EntryId = v.EntryId
+						(*bscData.BulkResponses)[index].Error = v.Error
 					}
 				}
-				filteredPubSubMsgs := utils.Filter(psm.pubSubMessages, func(ps message) bool {
-					if index, ok := (*bscData.entryIdIndexMap)[ps.entry.EntryId]; ok {
-						return (*bscData.bulkResponses)[index].Error != nil
+				filteredPubSubMsgs := utils.Filter(psm.PubSubMessages, func(ps todo.Message) bool {
+					if index, ok := (*bscData.EntryIdIndexMap)[ps.Entry.EntryId]; ok {
+						return (*bscData.BulkResponses)[index].Error != nil
 					}
 					return false
 				})
-				psm.pubSubMessages = filteredPubSubMsgs
-				psm.length = len(filteredPubSubMsgs)
+				psm.PubSubMessages = filteredPubSubMsgs
+				psm.Length = len(filteredPubSubMsgs)
 			},
 		})
-	_, err := policyRunner(func(ctx context.Context) (*bulkSubscribeResiliencyRes, error) {
-		var pErr error
-		bsrr := &bulkSubscribeResiliencyRes{
-			entries:  make([]contribpubsub.BulkSubscribeResponseEntry, 0, len(psm.pubSubMessages)),
-			envelope: maps.Clone(envelope),
+	_, err := policyRunner(func(ctx context.Context) (*todo.BulkSubscribeResiliencyRes, error) {
+		bsrr := &todo.BulkSubscribeResiliencyRes{
+			Entries:  make([]contribpubsub.BulkSubscribeResponseEntry, 0, len(psm.PubSubMessages)),
+			Envelope: maps.Clone(envelope),
 		}
-		if s.isHTTP {
-			pErr = s.publishBulkMessageHTTP(ctx, &bscData, &psm, bsrr, deadLetterTopic)
-		} else {
-			pErr = s.publishBulkMessageGRPC(ctx, &bscData, &psm, &bsrr.entries, rawPayload, deadLetterTopic)
-		}
-		return bsrr, pErr
+		err := s.postman.DeliverBulk(ctx, &postman.DelivererBulkRequest{
+			BulkSubCallData:      &bscData,
+			BulkSubMsg:           &psm,
+			BulkSubResiliencyRes: bsrr,
+			BulkResponses:        &bsrr.Entries,
+			RawPayload:           rawPayload,
+			DeadLetterTopic:      deadLetterTopic,
+		})
+		return bsrr, err
 	})
 	// setting error if any entry has not been yet touched - only use case that seems possible is of timeout
-	for eId, ind := range *bscData.entryIdIndexMap { //nolint:stylecheck
-		if (*bscData.bulkResponses)[ind].EntryId == "" {
-			(*bscData.bulkResponses)[ind].EntryId = eId
-			(*bscData.bulkResponses)[ind].Error = err
+	for eId, ind := range *bscData.EntryIdIndexMap { //nolint:stylecheck
+		if (*bscData.BulkResponses)[ind].EntryId == "" {
+			(*bscData.BulkResponses)[ind].EntryId = eId
+			(*bscData.BulkResponses)[ind].Error = err
 		}
 	}
-	return bscData.bulkResponses, err
+	return bscData.BulkResponses, err
 }