SY-4333: Move Signal Propagation Out of Distribution Layer

core/pkg/distribution/channel/create_test.go

@@ -146,7 +146,7 @@ var _ = Describe("Create", Ordered, func() {
 			})
 			It("Should create the channel without error", func(ctx SpecContext) {
 				Expect(ch.Key().Leaseholder()).To(Equal(aspen.NodeKeyFree))
-				Expect(ch.Key().LocalKey()).To(Equal(channel.LocalKey(5)))
+				Expect(ch.Key().LocalKey()).To(Equal(channel.LocalKey(1)))
 				Expect(mockCluster.Nodes[1].Storage.TS.RetrieveChannels(ctx, ch.Key().StorageKey())).
 					Error().To(MatchError(query.ErrNotFound))
 			})

core/pkg/distribution/layer.go

@@ -12,22 +12,17 @@ package distribution
 import (
 	"context"
 	"fmt"
-	"io"
 
 	"github.com/samber/lo"
 	"github.com/synnaxlabs/alamos"
 	"github.com/synnaxlabs/aspen"
 	"github.com/synnaxlabs/synnax/pkg/distribution/channel"
-	channelsignals "github.com/synnaxlabs/synnax/pkg/distribution/channel/signals"
 	"github.com/synnaxlabs/synnax/pkg/distribution/channel/verification"
 	"github.com/synnaxlabs/synnax/pkg/distribution/framer"
 	"github.com/synnaxlabs/synnax/pkg/distribution/group"
-	groupsignals "github.com/synnaxlabs/synnax/pkg/distribution/group/signals"
 	"github.com/synnaxlabs/synnax/pkg/distribution/node"
 	"github.com/synnaxlabs/synnax/pkg/distribution/ontology"
-	ontologysignals "github.com/synnaxlabs/synnax/pkg/distribution/ontology/signals"
 	"github.com/synnaxlabs/synnax/pkg/distribution/search"
-	"github.com/synnaxlabs/synnax/pkg/distribution/signals"
 	"github.com/synnaxlabs/synnax/pkg/storage"
 	"github.com/synnaxlabs/x/address"
 	"github.com/synnaxlabs/x/config"
@@ -75,11 +70,6 @@ type LayerConfig struct {
 	//
 	// [REQUIRED]
 	Storage *storage.Layer
-	// EnableServiceSignals sets whether to enable CDC signal propagation for changes
-	// to distribution layer data structures (channels, groups, etc.)
-	//
-	// [OPTIONAL] - Defaults to true.
-	EnableServiceSignals *bool
 	// ValidateChannelNames disables channel name validation when true.
 	// This allows channels with special characters, spaces, etc.
 	//
@@ -116,7 +106,6 @@ var (
 	// required fields specific in Config.
 	DefaultLayerConfig = LayerConfig{
 		GorpCodec:            orc.NewCodec(msgpack.Codec),
-		EnableServiceSignals: new(true),
 		ValidateChannelNames: new(true),
 	}
 )
@@ -134,7 +123,6 @@ func (c LayerConfig) Override(other LayerConfig) LayerConfig {
 	c.Verifier = override.String(c.Verifier, other.Verifier)
 	c.TestingIntOverflowCheck = override.Nil(c.TestingIntOverflowCheck, other.TestingIntOverflowCheck)
 	c.GorpCodec = override.Nil(c.GorpCodec, other.GorpCodec)
-	c.EnableServiceSignals = override.Nil(c.EnableServiceSignals, other.EnableServiceSignals)
 	c.ValidateChannelNames = override.Nil(c.ValidateChannelNames, other.ValidateChannelNames)
 	return c
 }
@@ -149,7 +137,6 @@ func (c LayerConfig) Validate() error {
 	validate.NotNil(v, "frame_transport", c.FrameTransport)
 	validate.NotNil(v, "aspen_transport", c.AspenTransport)
 	validate.NotNil(v, "codec", c.GorpCodec)
-	validate.NotNil(v, "enable_channel_signals", c.EnableServiceSignals)
 	validate.NotNil(v, "disable_channel_name_validation", c.ValidateChannelNames)
 	return v.Error()
 }
@@ -179,9 +166,6 @@ type Layer struct {
 	// Search is the full-text search index for ontology resources.
 	// [REQUIRED]
 	Search *search.Index
-	// Signals are for propagating changes to data structures through channels in
-	// Synnax.
-	Signals *signals.Provider
 	// Group is for grouping related resources in the cluster.
 	Group *group.Service
 	// Verification verifies that the universe remains as it is.
@@ -302,40 +286,6 @@ func OpenLayer(ctx context.Context, cfgs ...LayerConfig) (l *Layer, err error) {
 		return nil, err
 	}
 
-	if l.Signals, err = signals.New(signals.Config{
-		Channel:         l.Channel,
-		Framer:          l.Framer,
-		Instrumentation: cfg.Child("signals"),
-	}); !ok(err, nil) {
-		return nil, err
-	}
-
-	if *cfg.EnableServiceSignals {
-		var channelSignalsCloser io.Closer
-		if channelSignalsCloser, err = channelsignals.Publish(
-			ctx,
-			l.Signals,
-			l.Channel.Observe(),
-		); !ok(err, channelSignalsCloser) {
-			return nil, err
-		}
-		var groupSignalsCloser io.Closer
-		if groupSignalsCloser, err = groupsignals.Publish(ctx, l.Signals, l.Group.Observe()); !ok(err, groupSignalsCloser) {
-			return nil, err
-		}
-	}
-
-	if l.Cluster.HostKey() == node.KeyBootstrapper {
-		var ontologyCDCCloser io.Closer
-		if ontologyCDCCloser, err = ontologysignals.Publish(
-			ctx,
-			l.Signals,
-			l.Ontology,
-		); !ok(err, ontologyCDCCloser) {
-			return nil, err
-		}
-	}
-
 	return l, err
 }
 

core/pkg/distribution/mock/cluster.go

@@ -102,7 +102,6 @@ func (c *Cluster) Provision(
 			AspenOptions: []aspen.Option{
 				aspen.WithPropagationConfig(aspen.FastPropagationConfig),
 			},
-			EnableServiceSignals: new(false),
 		}, c.cfg}, cfgs...)...))
 	)
 	node := Node{Layer: distributionLayer, Storage: storageLayer}

core/pkg/distribution/ontology/service.go

@@ -31,9 +31,9 @@ type Service interface {
 	// RetrieveResource returns the resource with the give key (Name.Name). If the resource
 	// does not exist, returns a query.ErrNotFound error.
 	RetrieveResource(ctx context.Context, key string, tx gorp.Tx) (Resource, error)
-	// Observable is used by the ontology to subscribe to changes in the entities.
-	// This is used to propagate changes via the ResourceObserver for CDC signals.
-	// If the service's entities are static, use observe.Noop.
+	// Observable is used by the ontology to subscribe to changes in the entities. This
+	// is used to propagate changes via the ResourceObserver for signals. If the
+	// service's entities are static, use observe.Noop.
 	observe.Observable[iter.Seq[Change]]
 }
 

core/pkg/service/access/rbac/policy/service.go

@@ -16,8 +16,8 @@ import (
 	"github.com/synnaxlabs/alamos"
 	"github.com/synnaxlabs/synnax/pkg/distribution/ontology"
 	"github.com/synnaxlabs/synnax/pkg/distribution/search"
-	"github.com/synnaxlabs/synnax/pkg/distribution/signals"
 	"github.com/synnaxlabs/synnax/pkg/service/access/rbac/policy/migrations/v0"
+	"github.com/synnaxlabs/synnax/pkg/service/signals"
 	"github.com/synnaxlabs/x/config"
 	"github.com/synnaxlabs/x/gorp"
 	xio "github.com/synnaxlabs/x/io"

core/pkg/service/access/rbac/role/service.go

@@ -17,7 +17,7 @@ import (
 	"github.com/synnaxlabs/synnax/pkg/distribution/group"
 	"github.com/synnaxlabs/synnax/pkg/distribution/ontology"
 	"github.com/synnaxlabs/synnax/pkg/distribution/search"
-	"github.com/synnaxlabs/synnax/pkg/distribution/signals"
+	"github.com/synnaxlabs/synnax/pkg/service/signals"
 	"github.com/synnaxlabs/x/config"
 	"github.com/synnaxlabs/x/gorp"
 	xio "github.com/synnaxlabs/x/io"

core/pkg/service/access/rbac/service.go

@@ -17,12 +17,12 @@ import (
 	"github.com/synnaxlabs/synnax/pkg/distribution/group"
 	"github.com/synnaxlabs/synnax/pkg/distribution/ontology"
 	"github.com/synnaxlabs/synnax/pkg/distribution/search"
-	"github.com/synnaxlabs/synnax/pkg/distribution/signals"
 	"github.com/synnaxlabs/synnax/pkg/service/access"
 	"github.com/synnaxlabs/synnax/pkg/service/access/rbac/builtin"
 	v0 "github.com/synnaxlabs/synnax/pkg/service/access/rbac/migrations/v0"
 	"github.com/synnaxlabs/synnax/pkg/service/access/rbac/policy"
 	"github.com/synnaxlabs/synnax/pkg/service/access/rbac/role"
+	"github.com/synnaxlabs/synnax/pkg/service/signals"
 	"github.com/synnaxlabs/synnax/pkg/service/user"
 	"github.com/synnaxlabs/x/config"
 	"github.com/synnaxlabs/x/errors"

core/pkg/service/actions/actions_suite_test.go

@@ -15,6 +15,7 @@ import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	"github.com/synnaxlabs/synnax/pkg/distribution/mock"
+	"github.com/synnaxlabs/synnax/pkg/service/signals"
 	. "github.com/synnaxlabs/x/testutil"
 )
 
@@ -25,11 +26,18 @@ func TestActions(t *testing.T) {
 
 var _ = ShouldNotLeakGoroutinesPerSpec()
 
-var dist mock.Node
+var (
+	dist mock.Node
+	sigs *signals.Provider
+)
 
 var _ = BeforeSuite(func(ctx SpecContext) {
 	cluster := DeferClose(mock.NewCluster())
 	dist = DeferClose(cluster.Provision(ctx))
+	sigs = MustSucceed(signals.New(signals.Config{
+		Channel: dist.Channel,
+		Framer:  dist.Framer,
+	}))
 })
 
 // testAction is a small concrete action type used to instantiate the generic

core/pkg/service/actions/signals.go

@@ -15,8 +15,8 @@ import (
 	"fmt"
 	"io"
 
-	"github.com/synnaxlabs/synnax/pkg/distribution/signals"
 	"github.com/synnaxlabs/synnax/pkg/service/channel"
+	"github.com/synnaxlabs/synnax/pkg/service/signals"
 	xchange "github.com/synnaxlabs/x/change"
 	"github.com/synnaxlabs/x/errors"
 	"github.com/synnaxlabs/x/observe"