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.

The existing implementation on subscription message processing and
routing to various sinks are desperate in the codebase and somewhat
spaghetti.

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

@@ -164,27 +164,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("subscription is closed")
+		return nil, errors.New("subscription 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()
 
@@ -209,35 +210,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
 }