fixup! add subscriber subsystem

coadler · coadler · commit 3af1af1f3560 · 2023-09-15T22:29:53.000Z
diff --git a/enterprise/tailnet/connio.go b/enterprise/tailnet/connio.go
@@ -34,7 +34,7 @@ func newConnIO(pCtx context.Context,
 	id uuid.UUID,
 	name string,
 	kind agpl.QueueKind,
-) (*connIO, error) {
+) *connIO {
 	ctx, cancel := context.WithCancel(pCtx)
 	c := &connIO{
 		pCtx:     pCtx,
@@ -48,7 +48,7 @@ func newConnIO(pCtx context.Context,
 	go c.recvLoop()
 	go c.updates.SendUpdates()
 	logger.Info(ctx, "serving connection")
-	return c, nil
+	return c
 }
 
 func (c *connIO) recvLoop() {
diff --git a/enterprise/tailnet/pgcoord.go b/enterprise/tailnet/pgcoord.go
@@ -26,16 +26,16 @@ import (
 )
 
 const (
-	EventHeartbeats         = "tailnet_coordinator_heartbeat"
-	eventClientUpdate       = "tailnet_client_update"
-	eventClientSubscription = "tailnet_client_subscription_update"
-	eventAgentUpdate        = "tailnet_agent_update"
-	HeartbeatPeriod         = time.Second * 2
-	MissedHeartbeats        = 3
-	numQuerierWorkers       = 10
-	numBinderWorkers        = 10
-	dbMaxBackoff            = 10 * time.Second
-	cleanupPeriod           = time.Hour
+	EventHeartbeats      = "tailnet_coordinator_heartbeat"
+	eventClientUpdate    = "tailnet_client_update"
+	eventAgentUpdate     = "tailnet_agent_update"
+	HeartbeatPeriod      = time.Second * 2
+	MissedHeartbeats     = 3
+	numQuerierWorkers    = 10
+	numBinderWorkers     = 10
+	numSubscriberWorkers = 10
+	dbMaxBackoff         = 10 * time.Second
+	cleanupPeriod        = time.Hour
 )
 
 // TODO: add subscriber to this graphic
@@ -240,10 +240,8 @@ func (c *pgCoord) ServeClient(conn net.Conn, id uuid.UUID, agent uuid.UUID) erro
 				slog.Error(err))
 		}
 	}()
-	cIO, err := newConnIO(c.ctx, c.logger, c.bindings, conn, id, id.String(), agpl.QueueKindClient)
-	if err != nil {
-		return err
-	}
+
+	cIO := newConnIO(c.ctx, c.logger, c.bindings, conn, id, id.String(), agpl.QueueKindClient)
 	if err := sendCtx(c.ctx, c.newConnections, agpl.Queue(cIO)); err != nil {
 		// can only be a context error, no need to log here.
 		return err
@@ -277,10 +275,7 @@ func (c *pgCoord) ServeAgent(conn net.Conn, id uuid.UUID, name string) error {
 		}
 	}()
 	logger := c.logger.With(slog.F("name", name))
-	cIO, err := newConnIO(c.ctx, logger, c.bindings, conn, id, name, agpl.QueueKindAgent)
-	if err != nil {
-		return err
-	}
+	cIO := newConnIO(c.ctx, logger, c.bindings, conn, id, name, agpl.QueueKindAgent)
 	if err := sendCtx(c.ctx, c.newConnections, agpl.Queue(cIO)); err != nil {
 		// can only be a context error, no need to log here.
 		return err
@@ -349,7 +344,7 @@ func newSubscriber(ctx context.Context,
 	go s.handleSubscriptions()
 	go func() {
 		<-startWorkers
-		for i := 0; i < numBinderWorkers; i++ {
+		for i := 0; i < numSubscriberWorkers; i++ {
 			go s.worker()
 		}
 	}()
diff --git a/tailnet/coordinator.go b/tailnet/coordinator.go
@@ -194,8 +194,7 @@ type core struct {
 type QueueKind int
 
 const (
-	_ QueueKind = iota
-	QueueKindClient
+	QueueKindClient QueueKind = 1 + iota
 	QueueKindAgent
 )
 

Original file line number	Diff line number	Diff line change
`@@ -194,8 +194,7 @@ type core struct {`
`194`	`194`	`type QueueKind int`
`195`	`195`
`196`	`196`	`const (`
`197`		`- _ QueueKind = iota`
`198`		`- QueueKindClient`
	`197`	`+ QueueKindClient QueueKind = 1 + iota`
`199`	`198`	`QueueKindAgent`
`200`	`199`	`)`
`201`	`200`