Added the notification channel: listeners woken on their scheduled task.

chasm/lib/channel/config.go

@@ -0,0 +1,138 @@
+package channel
+
+import (
+	"time"
+
+	"go.temporal.io/server/common/dynamicconfig"
+)
+
+// MaxPositionBytes bounds the position a notification carries. It is a cursor
+// for the listener to read from rather than data, and the channel retains it.
+const MaxPositionBytes = 1024
+
+// LongPollTimeout matches the other history long polls: on expiry the caller
+// gets an empty response and polls again.
+const LongPollTimeout = 20 * time.Second
+
+// LongPollBuffer leaves room to answer before the caller's own deadline.
+const LongPollBuffer = 3 * time.Second
+
+// RoutedCallTimeout bounds one call to another shard, so a slow listener does
+// not take the whole budget of the task that is telling the others.
+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
+)
+
+var (
+	MaxListenersSetting = dynamicconfig.NewNamespaceIntSetting(
+		"channel.maxListeners",
+		DefaultMaxListeners,
+		`Most listeners, workflow and callback together, one notification channel holds. A
+registration past it is refused.`,
+	)
+	RetainedNotificationsSetting = dynamicconfig.NewNamespaceIntSetting(
+		"channel.retainedNotifications",
+		DefaultRetainedNotifications,
+		`Most notifications a channel keeps for pollers. The oldest goes when a new one
+arrives past it.`,
+	)
+	MaxMetadataBytesSetting = dynamicconfig.NewNamespaceIntSetting(
+		"channel.maxMetadataBytes",
+		DefaultMaxMetadataBytes,
+		`Largest notification metadata accepted, summed over its payloads.`,
+	)
+	RetentionSetting = dynamicconfig.NewNamespaceDurationSetting(
+		"channel.retention",
+		DefaultRetention,
+		`How long a channel with no listeners stays after its last notify, registration or
+poll before it is deleted with what it retained.`,
+	)
+	NotifyPerSecondSetting = dynamicconfig.NewNamespaceIntSetting(
+		"channel.notifyPerSecond",
+		DefaultNotifyPerSecond,
+		`Most NotifyChannel calls a namespace may make per second on one history host. Zero
+means no limit.`,
+	)
+	MaxSubscriptionsPerWorkflowSetting = dynamicconfig.NewNamespaceIntSetting(
+		"channel.maxSubscriptionsPerWorkflow",
+		DefaultMaxSubscriptionsPerWorkflow,
+		`Most notification channels one workflow run subscribes to. A subscribe command past
+it fails the workflow task.`,
+	)
+)
+
+// Limits are the per-namespace bounds a channel transition applies, resolved
+// by the caller so the component needs no config of its own.
+type Limits struct {
+	MaxListeners          int
+	RetainedNotifications int
+	MaxMetadataBytes      int
+	Retention             time.Duration
+}
+
+// DefaultLimits is what applies when nothing is configured.
+func DefaultLimits() Limits {
+	return Limits{
+		MaxListeners:          DefaultMaxListeners,
+		RetainedNotifications: DefaultRetainedNotifications,
+		MaxMetadataBytes:      DefaultMaxMetadataBytes,
+		Retention:             DefaultRetention,
+	}
+}
+
+// withDefaults fills what a caller left at zero.
+func (l Limits) withDefaults() Limits {
+	d := DefaultLimits()
+	if l.MaxListeners <= 0 {
+		l.MaxListeners = d.MaxListeners
+	}
+	if l.RetainedNotifications <= 0 {
+		l.RetainedNotifications = d.RetainedNotifications
+	}
+	if l.MaxMetadataBytes <= 0 {
+		l.MaxMetadataBytes = d.MaxMetadataBytes
+	}
+	if l.Retention <= 0 {
+		l.Retention = d.Retention
+	}
+	return l
+}
+
+type Config struct {
+	MaxListeners                dynamicconfig.IntPropertyFnWithNamespaceFilter
+	RetainedNotifications       dynamicconfig.IntPropertyFnWithNamespaceFilter
+	MaxMetadataBytes            dynamicconfig.IntPropertyFnWithNamespaceFilter
+	Retention                   dynamicconfig.DurationPropertyFnWithNamespaceFilter
+	NotifyPerSecond             dynamicconfig.IntPropertyFnWithNamespaceFilter
+	MaxSubscriptionsPerWorkflow dynamicconfig.IntPropertyFnWithNamespaceFilter
+}
+
+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),
+	}
+}
+
+// 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),
+	}
+}

chasm/lib/channel/convert.go

@@ -0,0 +1,71 @@
+package channel
+
+import (
+	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.
+
+func FromAPINotification(n *notificationpb.Notification) *channelpb.Notification {
+	if n == nil {
+		return nil
+	}
+	n = common.CloneProto(n)
+	return &channelpb.Notification{
+		Channel:  n.GetChannel(),
+		Position: n.GetPosition(),
+		Counter:  n.GetCounter(),
+		Metadata: n.GetMetadata(),
+	}
+}
+
+func ToAPINotification(n *channelpb.Notification) *notificationpb.Notification {
+	if n == nil {
+		return nil
+	}
+	n = common.CloneProto(n)
+	return &notificationpb.Notification{
+		Channel:  n.GetChannel(),
+		Position: n.GetPosition(),
+		Counter:  n.GetCounter(),
+		Metadata: n.GetMetadata(),
+	}
+}
+
+func ToAPINotifications(ns []*channelpb.Notification) []*notificationpb.Notification {
+	if len(ns) == 0 {
+		return nil
+	}
+	out := make([]*notificationpb.Notification, len(ns))
+	for i, n := range ns {
+		out[i] = ToAPINotification(n)
+	}
+	return out
+}
+
+// ToAPIListener converts a listener from describe.
+func ToAPIListener(l *channelpb.ChannelListenerInfo) *notificationpb.ChannelListener {
+	out := &notificationpb.ChannelListener{
+		ListenerId:     l.GetListenerId(),
+		RegisteredTime: common.CloneProto(l.GetRegisteredTime()),
+	}
+	switch variant := l.GetVariant().(type) {
+	case *channelpb.ChannelListenerInfo_Workflow_:
+		out.Listener = &notificationpb.ChannelListener_Workflow{
+			Workflow: &notificationpb.WorkflowListener{
+				WorkflowId: variant.Workflow.GetWorkflowId(),
+				RunId:      variant.Workflow.GetRunId(),
+			},
+		}
+	case *channelpb.ChannelListenerInfo_Callback:
+		out.Listener = &notificationpb.ChannelListener_Callback{
+			Callback: common.CloneProto(variant.Callback),
+		}
+	default:
+		// A listener with no variant has nothing more to say.
+	}
+	return out
+}