Added the notification channel linked to a workflow.

chasm/lib/channel/channel.go

@@ -194,7 +194,7 @@ func (c *Channel) RetainedCount() int64 {
 }
 
 func (c *Channel) scheduleFanOut(mctx chasm.MutableContext) {
-	if c.State.FanOutPending || c.ListenerCount() == 0 {
+	if c.State.GetLinked() || c.State.FanOutPending || c.ListenerCount() == 0 {
 		return
 	}
 	c.State.FanOutPending = true
@@ -516,6 +516,8 @@ type CallbackDelivery struct {
 	Callback     *commonpb.Callback
 	Notification *channelpb.Notification
 	Attempt      int32
+	// The channel is linked to a workflow, for the metrics.
+	Linked bool
 }
 
 // CallbackDeliveryFor reads what the delivery task should post. It reports
@@ -538,6 +540,7 @@ func (c *Channel) CallbackDeliveryFor(
 		Callback:     common.CloneProto(listener.GetCallback()),
 		Notification: common.CloneProto(listener.GetInFlight()),
 		Attempt:      listener.GetAttempt(),
+		Linked:       c.State.GetLinked(),
 	}, true
 }
 
@@ -740,7 +743,7 @@ func (c *Channel) PollNeedsTouch(now time.Time, limits Limits) bool {
 // scheduleIdleCheck arms the deletion check for a channel with no listeners.
 // One is outstanding at a time; it re-arms itself while the channel is in use.
 func (c *Channel) scheduleIdleCheck(mctx chasm.MutableContext, limits Limits) {
-	if c.State.IdleCheckPending || c.ListenerCount() > 0 {
+	if c.State.GetLinked() || c.State.IdleCheckPending || c.ListenerCount() > 0 {
 		return
 	}
 	c.State.IdleCheckPending = true

chasm/lib/channel/config.go

@@ -24,12 +24,14 @@ const RoutedCallTimeout = 5 * time.Second
 // The defaults below are also what component code falls back on when it is
 // driven without a config, as the unit tests do.
 const (
-	DefaultMaxListeners                = 1000
-	DefaultRetainedNotifications       = 1000
-	DefaultMaxMetadataBytes            = 2048
-	DefaultRetention                   = time.Hour
-	DefaultNotifyPerSecond             = 10000
-	DefaultMaxSubscriptionsPerWorkflow = 32
+	DefaultMaxListeners                 = 1000
+	DefaultRetainedNotifications        = 1000
+	DefaultMaxMetadataBytes             = 2048
+	DefaultRetention                    = time.Hour
+	DefaultNotifyPerSecond              = 10000
+	DefaultMaxSubscriptionsPerWorkflow  = 32
+	DefaultLinkedRetainedNotifications  = 100
+	DefaultMaxLinkedChannelsPerWorkflow = 32
 )
 
 var (
@@ -68,6 +70,31 @@ means no limit.`,
 		`Most notification channels one workflow run subscribes to. A subscribe command past
 it fails the workflow task.`,
 	)
+	LinkedRetainedNotificationsSetting = dynamicconfig.NewNamespaceIntSetting(
+		"channel.linkedRetainedNotifications",
+		DefaultLinkedRetainedNotifications,
+		`Most notifications a channel linked to a workflow keeps for pollers. Held in the
+workflow's own state, so smaller than an independent channel's ring.`,
+	)
+	MaxLinkedChannelsPerWorkflowSetting = dynamicconfig.NewNamespaceIntSetting(
+		"channel.maxLinkedChannelsPerWorkflow",
+		DefaultMaxLinkedChannelsPerWorkflow,
+		`Most channels linked to one workflow run. A notify or a registration that would
+create one past it is refused with ResourceExhausted.`,
+	)
+	// LinkedKindEnabledSetting exists so the independent kind can be exercised
+	// live on a server that has the linked kind: a client that sends
+	// workflow_execution is routed to the independent channel of that name, as
+	// before the linked kind existed, and a probe lands on INDEPENDENT.
+	LinkedKindEnabledSetting = dynamicconfig.NewNamespaceBoolSetting(
+		"channel.linkedKindEnabled",
+		true,
+		`Whether the public channel calls honour workflow_execution and reach the channel
+linked to that workflow. Off, they ignore it and reach the independent channel of
+the name, so DescribeChannel on an untouched linked name answers NotFound. The
+channels a native stream drives in its owner's state are not affected: they are
+written by the server, not addressed through these calls.`,
+	)
 )
 
 // Limits are the per-namespace bounds a channel transition applies, resolved
@@ -77,15 +104,20 @@ type Limits struct {
 	RetainedNotifications int
 	MaxMetadataBytes      int
 	Retention             time.Duration
+	// The linked kind's ring and how many linked channels one run may hold.
+	LinkedRetainedNotifications  int
+	MaxLinkedChannelsPerWorkflow int
 }
 
 // DefaultLimits is what applies when nothing is configured.
 func DefaultLimits() Limits {
 	return Limits{
-		MaxListeners:          DefaultMaxListeners,
-		RetainedNotifications: DefaultRetainedNotifications,
-		MaxMetadataBytes:      DefaultMaxMetadataBytes,
-		Retention:             DefaultRetention,
+		MaxListeners:                 DefaultMaxListeners,
+		RetainedNotifications:        DefaultRetainedNotifications,
+		MaxMetadataBytes:             DefaultMaxMetadataBytes,
+		Retention:                    DefaultRetention,
+		LinkedRetainedNotifications:  DefaultLinkedRetainedNotifications,
+		MaxLinkedChannelsPerWorkflow: DefaultMaxLinkedChannelsPerWorkflow,
 	}
 }
 
@@ -104,35 +136,49 @@ func (l Limits) withDefaults() Limits {
 	if l.Retention <= 0 {
 		l.Retention = d.Retention
 	}
+	if l.LinkedRetainedNotifications <= 0 {
+		l.LinkedRetainedNotifications = d.LinkedRetainedNotifications
+	}
+	if l.MaxLinkedChannelsPerWorkflow <= 0 {
+		l.MaxLinkedChannelsPerWorkflow = d.MaxLinkedChannelsPerWorkflow
+	}
 	return l
 }
 
 type Config struct {
-	MaxListeners                dynamicconfig.IntPropertyFnWithNamespaceFilter
-	RetainedNotifications       dynamicconfig.IntPropertyFnWithNamespaceFilter
-	MaxMetadataBytes            dynamicconfig.IntPropertyFnWithNamespaceFilter
-	Retention                   dynamicconfig.DurationPropertyFnWithNamespaceFilter
-	NotifyPerSecond             dynamicconfig.IntPropertyFnWithNamespaceFilter
-	MaxSubscriptionsPerWorkflow dynamicconfig.IntPropertyFnWithNamespaceFilter
+	MaxListeners                 dynamicconfig.IntPropertyFnWithNamespaceFilter
+	RetainedNotifications        dynamicconfig.IntPropertyFnWithNamespaceFilter
+	MaxMetadataBytes             dynamicconfig.IntPropertyFnWithNamespaceFilter
+	Retention                    dynamicconfig.DurationPropertyFnWithNamespaceFilter
+	NotifyPerSecond              dynamicconfig.IntPropertyFnWithNamespaceFilter
+	MaxSubscriptionsPerWorkflow  dynamicconfig.IntPropertyFnWithNamespaceFilter
+	LinkedRetainedNotifications  dynamicconfig.IntPropertyFnWithNamespaceFilter
+	MaxLinkedChannelsPerWorkflow dynamicconfig.IntPropertyFnWithNamespaceFilter
+	LinkedKindEnabled            dynamicconfig.BoolPropertyFnWithNamespaceFilter
 }
 
 func NewConfig(dc *dynamicconfig.Collection) *Config {
 	return &Config{
-		MaxListeners:                MaxListenersSetting.Get(dc),
-		RetainedNotifications:       RetainedNotificationsSetting.Get(dc),
-		MaxMetadataBytes:            MaxMetadataBytesSetting.Get(dc),
-		Retention:                   RetentionSetting.Get(dc),
-		NotifyPerSecond:             NotifyPerSecondSetting.Get(dc),
-		MaxSubscriptionsPerWorkflow: MaxSubscriptionsPerWorkflowSetting.Get(dc),
+		LinkedKindEnabled:            LinkedKindEnabledSetting.Get(dc),
+		MaxListeners:                 MaxListenersSetting.Get(dc),
+		RetainedNotifications:        RetainedNotificationsSetting.Get(dc),
+		MaxMetadataBytes:             MaxMetadataBytesSetting.Get(dc),
+		Retention:                    RetentionSetting.Get(dc),
+		NotifyPerSecond:              NotifyPerSecondSetting.Get(dc),
+		MaxSubscriptionsPerWorkflow:  MaxSubscriptionsPerWorkflowSetting.Get(dc),
+		LinkedRetainedNotifications:  LinkedRetainedNotificationsSetting.Get(dc),
+		MaxLinkedChannelsPerWorkflow: MaxLinkedChannelsPerWorkflowSetting.Get(dc),
 	}
 }
 
 // LimitsFor resolves the namespace's limits.
 func (c *Config) LimitsFor(namespaceName string) Limits {
 	return Limits{
-		MaxListeners:          c.MaxListeners(namespaceName),
-		RetainedNotifications: c.RetainedNotifications(namespaceName),
-		MaxMetadataBytes:      c.MaxMetadataBytes(namespaceName),
-		Retention:             c.Retention(namespaceName),
+		MaxListeners:                 c.MaxListeners(namespaceName),
+		RetainedNotifications:        c.RetainedNotifications(namespaceName),
+		MaxMetadataBytes:             c.MaxMetadataBytes(namespaceName),
+		Retention:                    c.Retention(namespaceName),
+		LinkedRetainedNotifications:  c.LinkedRetainedNotifications(namespaceName),
+		MaxLinkedChannelsPerWorkflow: c.MaxLinkedChannelsPerWorkflow(namespaceName),
 	}
 }

chasm/lib/channel/convert.go

@@ -1,13 +1,17 @@
 package channel
 
 import (
+	commonpb "go.temporal.io/api/common/v1"
+	enumspb "go.temporal.io/api/enums/v1"
 	notificationpb "go.temporal.io/api/notification/v1"
 	channelpb "go.temporal.io/server/chasm/lib/channel/gen/channelpb/v1"
 	"go.temporal.io/server/common"
 )
 
 // The stored notification has the public message's fields under the same
-// names, so converting is a field copy in either direction.
+// names, so converting is a field copy in either direction. The owner of a
+// linked channel is the one exception: the public message names it as an
+// Execution, the stored one as the workflow run it is.
 
 func FromAPINotification(n *notificationpb.Notification) *channelpb.Notification {
 	if n == nil {
@@ -19,6 +23,7 @@ func FromAPINotification(n *notificationpb.Notification) *channelpb.Notification
 		Position: n.GetPosition(),
 		Counter:  n.GetCounter(),
 		Metadata: n.GetMetadata(),
+		LinkedTo: FromAPIExecution(n.GetLinkedTo()),
 	}
 }
 
@@ -32,6 +37,7 @@ func ToAPINotification(n *channelpb.Notification) *notificationpb.Notification {
 		Position: n.GetPosition(),
 		Counter:  n.GetCounter(),
 		Metadata: n.GetMetadata(),
+		LinkedTo: ToAPIExecution(n.GetLinkedTo()),
 	}
 }
 
@@ -46,6 +52,27 @@ func ToAPINotifications(ns []*channelpb.Notification) []*notificationpb.Notifica
 	return out
 }
 
+// ToAPIExecution names a workflow run as the public Execution.
+func ToAPIExecution(e *commonpb.WorkflowExecution) *commonpb.Execution {
+	if e == nil {
+		return nil
+	}
+	return &commonpb.Execution{
+		Type:       enumspb.EXECUTION_TYPE_WORKFLOW,
+		BusinessId: e.GetWorkflowId(),
+		RunId:      e.GetRunId(),
+	}
+}
+
+// FromAPIExecution reads an Execution as the workflow run it names. The type
+// is not checked here: only workflows hold linked channels.
+func FromAPIExecution(e *commonpb.Execution) *commonpb.WorkflowExecution {
+	if e == nil {
+		return nil
+	}
+	return &commonpb.WorkflowExecution{WorkflowId: e.GetBusinessId(), RunId: e.GetRunId()}
+}
+
 // ToAPIListener converts a listener from describe.
 func ToAPIListener(l *channelpb.ChannelListenerInfo) *notificationpb.ChannelListener {
 	out := &notificationpb.ChannelListener{