SY-4432: Refactor distribution proxy And transport (#2531)

Author: pjdotson

aspen/options.go

@@ -186,7 +186,7 @@ func mergeDefaultOptions(o *options) {
 	// provide their own transport. Otherwise the eagerly-created pool would
 	// leak: nothing would reference it and nothing would close it.
 	if o.transport.Transport == nil {
-		pool := fgrpc.NewPool("", grpc.WithTransportCredentials(insecure.NewCredentials()))
+		pool := fgrpc.OpenPool("", grpc.WithTransportCredentials(insecure.NewCredentials()))
 		o.transport.ownedPool = pool
 		o.transport.Transport = grpct.New(pool)
 	}

core/cmd/start/start.go

@@ -14,21 +14,18 @@ import (
 	"io"
 	"os"
 	"path/filepath"
-	"slices"
 	"strconv"
 	"time"
 
 	"github.com/samber/lo"
 	"github.com/synnaxlabs/alamos"
 	aspentransport "github.com/synnaxlabs/aspen/transport/grpc"
-	"github.com/synnaxlabs/freighter/grpc"
 	"github.com/synnaxlabs/freighter/http"
 	cmdcert "github.com/synnaxlabs/synnax/cmd/cert"
 	"github.com/synnaxlabs/synnax/pkg/api"
 	"github.com/synnaxlabs/synnax/pkg/console"
 	"github.com/synnaxlabs/synnax/pkg/distribution"
-	channeltransport "github.com/synnaxlabs/synnax/pkg/distribution/transport/grpc/channel"
-	framertransport "github.com/synnaxlabs/synnax/pkg/distribution/transport/grpc/framer"
+	disttransport "github.com/synnaxlabs/synnax/pkg/distribution/transport/grpc"
 	"github.com/synnaxlabs/synnax/pkg/driver"
 	"github.com/synnaxlabs/synnax/pkg/security"
 	"github.com/synnaxlabs/synnax/pkg/security/cert"
@@ -42,10 +39,10 @@ import (
 	"github.com/synnaxlabs/x/config"
 	"github.com/synnaxlabs/x/errors"
 	xio "github.com/synnaxlabs/x/io"
-	xfs "github.com/synnaxlabs/x/io/fs"
+	"github.com/synnaxlabs/x/io/fs"
 	"github.com/synnaxlabs/x/override"
 	xservice "github.com/synnaxlabs/x/service"
-	xsignal "github.com/synnaxlabs/x/signal"
+	"github.com/synnaxlabs/x/signal"
 	"github.com/synnaxlabs/x/validate"
 	"go.uber.org/zap"
 )
@@ -199,23 +196,16 @@ func BootupCore(ctx context.Context, onServerStarted chan struct{}, cfgs ...Core
 		return ctx.Err()
 	}
 	var (
-		aspenTransport         = aspentransport.New(grpcClientPool)
-		frameTransport         = framertransport.New(grpcClientPool)
-		channelTransport       = channeltransport.New(grpcClientPool)
-		distributionTransports = []grpc.BindableTransport{
-			aspenTransport,
-			frameTransport,
-			channelTransport,
-		}
+		aspenTransport = aspentransport.New(grpcClientPool)
+		distTransport  = disttransport.New(grpcClientPool)
 	)
 
 	if distributionLayer, err = distribution.OpenLayer(ctx, distribution.LayerConfig{
 		Instrumentation:      cfg.Child("distribution"),
 		AdvertiseAddress:     cfg.listenAddress,
 		PeerAddresses:        cfg.peers,
 		AspenTransport:       aspenTransport,
-		FrameTransport:       frameTransport,
-		ChannelTransport:     channelTransport,
+		Transport:            distTransport,
 		Verifier:             cfg.verifier,
 		Storage:              storageLayer,
 		ValidateChannelNames: cfg.validateChannelNames,
@@ -270,9 +260,10 @@ func BootupCore(ctx context.Context, onServerStarted chan struct{}, cfgs ...Core
 				&server.SecureHTTPBranch{
 					Transports: []http.BindableTransport{r, embeddedConsole},
 				},
-				&server.GRPCBranch{Transports: slices.Concat(
+				&server.GRPCBranch{Transports: append(
 					transportLayer.GRPC,
-					distributionTransports,
+					aspenTransport,
+					distTransport,
 				)},
 				server.NewHTTPRedirectBranch(),
 			},
@@ -350,7 +341,7 @@ func openWorkDir() (string, io.Closer, error) {
 		"workdir",
 		strconv.Itoa(os.Getpid()),
 	)
-	if err = os.MkdirAll(dir, xfs.UserRWX); err != nil {
+	if err = os.MkdirAll(dir, fs.UserRWX); err != nil {
 		return "", nil, err
 	}
 	return dir, xio.CloserFunc(func() error { return os.RemoveAll(dir) }), nil
@@ -363,12 +354,12 @@ func runStartupSearchIndexing(
 	// Run indexing inside an isolated signal context, so that if we receive an early
 	// cancellation signal, we can ensure that we exit indexing before we close any
 	// resources that it depends on (notably storage KV).
-	searchIndexCtx, cancelIndexing := xsignal.WithCancel(ctx)
+	searchIndexCtx, cancelIndexing := signal.WithCancel(ctx)
 	searchIndexCtx.Go(
 		dist.Search.Initialize,
-		xsignal.WithKey("startup_search_indexing"),
+		signal.WithKey("startup_search_indexing"),
 	)
-	return xsignal.NewHardShutdown(searchIndexCtx, cancelIndexing)
+	return signal.NewHardShutdown(searchIndexCtx, cancelIndexing)
 }
 
 func parseIntegrations(enabled, disabled []string) []string {

core/cmd/start/transport.go

@@ -22,7 +22,7 @@ func configureClientGRPC(
 	sec security.Provider,
 	insecure bool,
 ) *fgrpc.Pool {
-	return fgrpc.NewPool(
+	return fgrpc.OpenPool(
 		"",
 		grpc.WithTransportCredentials(getClientGRPCTransportCredentials(sec, insecure)),
 	)

core/pkg/distribution/channel/pb/services.pb.go

@@ -16,33 +16,61 @@ import "google/protobuf/empty.proto";
 
 option go_package = "github.com/synnaxlabs/synnax/pkg/distribution/channel/pb";
 
-service ChannelCreateService {
+// CreateService forwards channel create requests from a gateway node to the leaseholder
+// that will own the new channels.
+service CreateService {
+  // Exec creates the requested channels and returns them populated with their assigned
+  // keys.
   rpc Exec(CreateMessage) returns (CreateMessage) {}
 }
 
-service ChannelDeleteService {
+// DeleteService forwards channel delete requests from a gateway node to the leaseholder
+// that owns the channels.
+service DeleteService {
+  // Exec deletes the requested channels.
   rpc Exec(DeleteRequest) returns (google.protobuf.Empty) {}
 }
 
-service ChannelRenameService {
+// RenameService forwards channel rename requests from a gateway node to the leaseholder
+// that owns the channels.
+service RenameService {
+  // Exec renames the requested channels.
   rpc Exec(RenameRequest) returns (google.protobuf.Empty) {}
 }
 
+// CreateOptions controls how a create request behaves when a channel with the same name
+// already exists.
 message CreateOptions {
+  // retrieve_if_name_exists returns the existing channel instead of creating a new one
+  // when a channel with the same name already exists.
   bool retrieve_if_name_exists = 1;
+  // overwrite_if_name_exists overwrites the existing channel when a channel with the
+  // same name already exists.
   bool overwrite_if_name_exists = 2;
 }
 
+// CreateMessage is the request and response payload for a channel create operation. On
+// the request the channels carry no keys; on the response they are populated with their
+// assigned keys.
 message CreateMessage {
+  // channels are the channels to create on the request, and the created channels with
+  // their assigned keys on the response.
   repeated pb.Channel channels = 1;
+  // opts controls create behavior on name collisions.
   CreateOptions opts = 2;
 }
 
+// DeleteRequest is the payload for a channel delete operation.
 message DeleteRequest {
+  // keys identifies the channels to delete.
   repeated uint32 keys = 1;
 }
 
+// RenameRequest is the payload for a channel rename operation. keys and names are
+// positional: the channel at keys[i] is renamed to names[i].
 message RenameRequest {
+  // keys identifies the channels to rename.
   repeated uint32 keys = 1;
+  // names holds the new name for each channel in keys, by position.
   repeated string names = 2;
 }