v0.21.0 scram password auth, cred and token management, browser tls-terminating proxy, viewer client

internal/sink/tcpchain/tcpchain.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/tls"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"maps"
 	"net"
@@ -50,7 +51,6 @@ type TCPChainSink struct {
 
 	node      string
 	addr      string
-	helloLine []byte
 	tlsConfig *tls.Config
 
 	// Authorization: pins the downstream server's identity
@@ -59,8 +59,12 @@ type TCPChainSink struct {
 	input  chan core.TransportEvent
 	logger *log.Logger
 
-	// conn owned exclusively by run loop goroutine
+	// conn and the backoff state are owned exclusively by the run loop goroutine.
+	// failures survives a link that dies right after connecting, so a refusing
+	// source is retried with growing delays rather than in a tight loop.
 	conn          net.Conn
+	connectedAt   time.Time
+	failures      int
 	everConnected bool
 	dialTimeout   time.Duration
 	writeTimeout  time.Duration
@@ -125,15 +129,11 @@ func NewTCPChainSinkPlugin(
 		}
 	}
 
-	helloLine, err := chain.EncodeHello(node)
-	if err != nil {
-		return nil, fmt.Errorf("hello: %w", err)
-	}
 	tlsCfg, err := tlsx.Client(opts.TLS, opts.Host)
 	if err != nil {
 		return nil, err
 	}
-	authPolicy, err := authz.New(opts.Auth, opts.TLS, authz.RoleDialer)
+	authPolicy, err := authz.New(opts.Auth, tlsCfg, authz.RoleDialer, authz.TCP)
 	if err != nil {
 		return nil, err
 	}
@@ -149,7 +149,6 @@ func NewTCPChainSinkPlugin(
 		config:       opts,
 		node:         node,
 		addr:         net.JoinHostPort(opts.Host, strconv.FormatInt(opts.Port, 10)),
-		helloLine:    helloLine,
 		tlsConfig:    tlsCfg,
 		auth:         authPolicy,
 		input:        make(chan core.TransportEvent, opts.BufferSize),
@@ -179,6 +178,7 @@ func NewTCPChainSinkPlugin(
 		"mtls", tlsCfg != nil && len(tlsCfg.Certificates) > 0,
 		"auth", authPolicy.Describe())
 	tlsx.LogWarnings(logger, "tcp_chain_sink", id, opts.TLS, false)
+	authPolicy.LogStartup(logger, "tcp_chain_sink", id, false)
 	return t, nil
 }
 
@@ -267,6 +267,17 @@ func (t *TCPChainSink) runLoop(ctx context.Context) {
 	defer t.wg.Done()
 	defer t.closeConn()
 
+	// Fold done into the context, so a connect or exchange in flight ends on Stop
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	go func() {
+		select {
+		case <-t.done:
+			cancel()
+		case <-ctx.Done():
+		}
+	}()
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -302,22 +313,30 @@ func (t *TCPChainSink) runLoop(ctx context.Context) {
 // deliver writes one line, holding it across reconnects until sent or shutdown.
 // Backpressure during outage propagates to the pipeline dispatch drop counter.
 func (t *TCPChainSink) deliver(ctx context.Context, line []byte) bool {
-	failures := 0
 	for {
 		if t.conn == nil {
-			if failures > 0 && !t.waitBackoff(ctx, failures) {
+			if t.failures > 0 && !t.waitBackoff(ctx, t.failures) {
 				return false
 			}
 			if err := t.connect(ctx); err != nil {
 				if ctx.Err() != nil {
 					return false
 				}
-				failures++
-				t.logger.Debug("msg", "Chain connect failed",
-					"component", "tcp_chain_sink",
-					"target", t.addr,
-					"attempt", failures,
-					"error", err)
+				t.failures++
+				if errors.Is(err, authz.ErrRefused) {
+					// A refusal is configuration, not weather: show it by default
+					t.logger.Warn("msg", "Chain connect refused",
+						"component", "tcp_chain_sink",
+						"target", t.addr,
+						"attempt", t.failures,
+						"error", err)
+				} else {
+					t.logger.Debug("msg", "Chain connect failed",
+						"component", "tcp_chain_sink",
+						"target", t.addr,
+						"attempt", t.failures,
+						"error", err)
+				}
 				continue
 			}
 		}
@@ -325,7 +344,7 @@ func (t *TCPChainSink) deliver(ctx context.Context, line []byte) bool {
 		t.conn.SetWriteDeadline(time.Now().Add(t.writeTimeout))
 		if _, err := t.conn.Write(line); err != nil {
 			t.writeErrors.Add(1)
-			failures++
+			t.failures++
 			t.logger.Warn("msg", "Chain write failed",
 				"component", "tcp_chain_sink",
 				"target", t.addr,
@@ -333,6 +352,10 @@ func (t *TCPChainSink) deliver(ctx context.Context, line []byte) bool {
 			t.closeConn()
 			continue
 		}
+		// Healthy once it outlives the shortest backoff
+		if t.failures > 0 && time.Since(t.connectedAt) >= time.Duration(t.config.BackoffMinMS)*time.Millisecond {
+			t.failures = 0
+		}
 		return true
 	}
 }
@@ -363,11 +386,26 @@ func (t *TCPChainSink) connect(ctx context.Context) error {
 		return err
 	}
 
-	conn.SetWriteDeadline(time.Now().Add(t.writeTimeout))
-	if _, err := conn.Write(t.helloLine); err != nil {
+	r, err := t.auth.Greet(ctx, conn, t.node)
+	if err != nil {
 		conn.Close()
-		return fmt.Errorf("hello: %w", err)
+		return err
 	}
+	// A source never writes once a link is up: a line is a refusal this sink
+	// could not otherwise see, EOF a close. Stop writing into either.
+	t.wg.Add(1)
+	go func() {
+		defer t.wg.Done()
+		err := authz.AwaitClose(r)
+		conn.Close()
+		if errors.Is(err, authz.ErrRefused) {
+			t.logger.Warn("msg", "Chain link refused",
+				"component", "tcp_chain_sink",
+				"target", t.addr,
+				"error", err)
+		}
+	}()
+	t.connectedAt = time.Now()
 
 	t.conn = conn
 	t.connected.Store(true)