config.go RawHistoryFolder

1package brokeripc
2
3import (
4	"time"
5
6	"github.com/bnema/vev/internal/adapters/brokerwire"
7	"github.com/bnema/vev/internal/protocol"
8)
9
10// Bounds. Every limit below is mandatory: no listener, session, or stream
11// grows an unbounded queue, and every default is small enough that a stalled
12// peer is refused rather than absorbed.
13const (
14	// DefaultMaxClients bounds concurrent accepted client connections on one
15	// listener. A listener acquires one slot before accepting, so the bound is
16	// enforced before a socket is admitted rather than after.
17	DefaultMaxClients = 64
18	// DefaultStreamInboundChunks bounds one logical stream's inbound chunk
19	// queue: frames already accepted from the wire and not yet consumed by the
20	// local session carriage.
21	DefaultStreamInboundChunks = 64
22	// DefaultStreamInboundBytes bounds one logical stream's inbound chunk queue
23	// in bytes (4 MiB).
24	DefaultStreamInboundBytes = 4 << 20
25)
26
27// Config bounds one endpoint. The zero value is valid and selects every
28// default; a negative or zero explicit bound is replaced by its default.
29type Config struct {
30	// MaxClients bounds concurrent accepted client connections.
31	MaxClients int
32	// StreamInboundChunks bounds one logical stream's inbound chunk queue.
33	StreamInboundChunks int
34	// StreamInboundBytes bounds one logical stream's inbound chunk bytes.
35	StreamInboundBytes uint64
36	// HandshakeTimeout bounds one accepted connection's broker preamble, broker
37	// admission, and the wait for its Register. The client dialer also bounds
38	// its whole setup (Unix dial, preamble, Register send, and wait for
39	// Registered) with it. On the server the registration phase shares the same
40	// accept-time budget as the preamble and admission: a peer that completes
41	// the preamble and admission but never sends Register is settled when that
42	// deadline elapses, so it cannot hold a core client lease or a listener slot
43	// indefinitely. It is deliberately separate from the session handshake
44	// budget that runs inside a logical stream.
45	HandshakeTimeout time.Duration
46	// MaxPendingOperations bounds mutating operations this side is waiting for
47	// on one connection. It never exceeds the brokerwire per-connection bound.
48	MaxPendingOperations int
49	// PassiveSubscribe marks every snapshot subscription of a client
50	// connection as a read-only reader: it receives publications but never
51	// counts as observation demand, so a CLI read never triggers a probe.
52	PassiveSubscribe bool
53	Build            string
54	OnRetire         func()
55}
56
57// withDefaults fills every unset bound with its package default.
58func (c Config) withDefaults() Config {
59	if c.Build == "" {
60		c.Build = defaultBuildIdentity()
61	}
62	if c.MaxClients <= 0 {
63		c.MaxClients = DefaultMaxClients
64	}
65	if c.StreamInboundChunks <= 0 {
66		c.StreamInboundChunks = DefaultStreamInboundChunks
67	}
68	if c.StreamInboundBytes == 0 {
69		c.StreamInboundBytes = DefaultStreamInboundBytes
70	}
71	if c.HandshakeTimeout <= 0 {
72		c.HandshakeTimeout = protocol.HandshakeTimeout
73	}
74	if c.MaxPendingOperations <= 0 {
75		c.MaxPendingOperations = brokerwire.MaxPendingOperations
76	}
77	return c
78}
79
80// validate refuses a configuration whose explicit bounds cannot be honored.
81// The zero value is always valid because withDefaults has already been applied.
82func (c Config) validate() error {
83	if c.MaxClients <= 0 || c.StreamInboundChunks <= 0 || c.StreamInboundBytes == 0 || c.HandshakeTimeout <= 0 {
84		return ErrConfig
85	}
86	if c.MaxPendingOperations <= 0 || c.MaxPendingOperations > brokerwire.MaxPendingOperations {
87		return ErrConfig
88	}
89	// A stream inbound byte bound below one maximum chunk can never accept a
90	// full stream frame, which would starve a legal peer.
91	if c.StreamInboundBytes < brokerwire.MaxStreamChunkBytes {
92		return ErrConfig
93	}
94	return nil
95}