1package daemonmux
2
3// Stateless bound refusals.
4//
5// One physical daemon connection negotiates one immutable set of ceilings
6// exactly once, in the daemonmux physical preamble: a receive envelope
7// ceiling (1 MiB..16 MiB), a stream chunk ceiling (1..64 KiB), a stream
8// count ceiling (1..128), and an aggregate buffered-bytes ceiling
9// (MinMuxAggregateBytes..64 MiB, one maximum chunk plus envelope overhead at
10// the floor). Every application envelope and every opaque data frame is
11// bounded statelessly before it allocates; the strictly-increasing stream
12// ID rule and the aggregate accounting are state-layer concerns owned by
13// the later engine slice, so this codec only enforces nonzero stream IDs.
14//
15// The daemonmux conversation carries no bulk category: every application
16// envelope is control-class and never exceeds 1 MiB, so one envelope ceiling
17// and one chunk ceiling cover the whole conversation. The negotiated receive
18// ceiling still bounds the physical carriage's own envelope reads, which may
19// sit above what a mux application envelope can ever need.
20
21import (
22 "errors"
23
24 "github.com/bnema/vev/internal/protocol/wire"
25)
26
27const (
28 // MuxRoleClient is the daemonmux preamble client role: the broker side
29 // that owns the physical connection. It is disjoint from the session
30 // roles 1/2 and the broker roles 3/4.
31 MuxRoleClient uint32 = 5
32 // MuxRoleServer is the daemonmux preamble server role: the daemon that
33 // accepts the multiplexed streams.
34 MuxRoleServer uint32 = 6
35
36 // MuxPreambleLimit bounds the serialized daemonmux preamble. It is the
37 // same 4 KiB bound the session and broker preambles use.
38 MuxPreambleLimit = 4 << 10
39
40 // MinMuxEnvelopeBytes is the advertised-envelope floor: a daemonmux peer
41 // must accept at least 1 MiB per application envelope
42 // (wire.ControlEnvelopeLimit, the control-class ceiling).
43 MinMuxEnvelopeBytes uint64 = wire.ControlEnvelopeLimit
44 // MaxMuxEnvelopeBytes is the advertised-envelope ceiling: a daemonmux
45 // peer must never advertise above 16 MiB (wire.AbsoluteEnvelopeLimit).
46 MaxMuxEnvelopeBytes uint64 = wire.AbsoluteEnvelopeLimit
47
48 // MaxMuxApplicationBytes is the absolute bound on one daemonmux
49 // application envelope. Every mux envelope is control-class metadata, so
50 // no application envelope ever exceeds 1 MiB (wire.ControlEnvelopeLimit)
51 // even when the negotiated receive ceiling admits a larger carriage
52 // envelope. The codec therefore enforces both bounds at once.
53 MaxMuxApplicationBytes uint64 = wire.ControlEnvelopeLimit
54
55 // MinMuxChunkBytes is the smallest legal stream chunk ceiling.
56 MinMuxChunkBytes uint64 = 1
57 // MaxMuxChunkBytes is the largest legal stream chunk ceiling: 64 KiB per
58 // opaque data frame.
59 MaxMuxChunkBytes uint64 = 64 << 10
60
61 // MinMuxStreams is the smallest legal concurrent-stream ceiling.
62 MinMuxStreams uint64 = 1
63 // MaxMuxStreams is the largest legal concurrent-stream ceiling.
64 MaxMuxStreams uint64 = 128
65
66 // MuxEnvelopeOverheadBytes is the largest carriage overhead one daemonmux
67 // data frame adds around its opaque payload: the outer directional oneof
68 // tag and length plus the MuxData field tags, the varint physical stream
69 // ID, and the varint payload length. The scheduler charges a frame's full
70 // serialized envelope against the aggregate byte budget, so an aggregate
71 // ceiling below one maximum chunk plus this overhead could refuse even a
72 // single legal frame. 64 bytes is a deliberate upper bound on that
73 // overhead (the worst case is under 32 bytes).
74 MuxEnvelopeOverheadBytes uint64 = 64
75
76 // MinMuxAggregateBytes is the smallest legal aggregate buffered-bytes
77 // ceiling: one maximum stream chunk plus the envelope overhead, so a peer
78 // that advertises the floor can always buffer at least one full-size
79 // frame in the direction it accepts.
80 MinMuxAggregateBytes uint64 = MaxMuxChunkBytes + MuxEnvelopeOverheadBytes
81 // MaxMuxAggregateBytes is the largest legal aggregate buffered-bytes
82 // ceiling: 64 MiB across every live stream of one physical connection.
83 MaxMuxAggregateBytes uint64 = 64 << 20
84
85 // MuxChunkCreditOverhead is the fixed credit one Data frame costs on top of
86 // its payload bytes. Charging it keeps a burst of tiny frames bounded in
87 // count as well as in bytes.
88 MuxChunkCreditOverhead uint64 = MuxEnvelopeOverheadBytes
89 // MaxMuxStreamWindowBytes caps one stream's receive window and every
90 // WindowUpdate grant.
91 MaxMuxStreamWindowBytes uint64 = 4 << 20
92)
93
94// chunkCredit is the flow-control cost of one Data frame of n payload bytes.
95func chunkCredit(n int) uint64 { return uint64(n) + MuxChunkCreditOverhead }
96
97// StreamWindow is the per-stream receive window both peers derive from the
98// same negotiated ceilings: the aggregate budget shared evenly across the
99// stream ceiling, never below one maximum chunk (a full chunk must always be
100// sendable) and never above MaxMuxStreamWindowBytes. With the default
101// ceilings it is 512 KiB. EffectiveMuxCeilings lowers the stream ceiling so
102// the one-chunk floor never pushes the sum of every live stream's window past
103// the aggregate: a conforming peer then never exceeds the aggregate budget.
104func (c MuxCeilings) StreamWindow() uint64 {
105 window := c.MaxAggregateBytes / max(c.MaxStreams, 1)
106 window = max(window, chunkCredit(int(c.StreamChunkLimit)))
107 return min(window, MaxMuxStreamWindowBytes)
108}
109
110// creditReturnThreshold is the consumed credit a receiver batches before it
111// grants a WindowUpdate: half the window, lowered when needed so a sender
112// blocked on a maximum chunk is always unblocked. A sender waits only once it
113// spent more than window-chunkCredit(limit), so granting at or below that
114// point can never deadlock.
115func (c MuxCeilings) creditReturnThreshold() uint64 {
116 window := c.StreamWindow()
117 cost := chunkCredit(int(c.StreamChunkLimit))
118 return max(min(window/2, window-cost+1), 1)
119}
120
121// ErrInvalidCeilings reports an advertisement outside its window: a zero,
122// below-floor, or above-ceiling envelope, chunk, stream-count, or aggregate
123// bound. The preamble maps it to its own ErrPreambleRejected sentinel; the
124// engine and scheduler constructors return it directly.
125var ErrInvalidCeilings = errors.New("daemonmux: invalid ceilings")
126
127// MuxCeilings is the validated, immutable set of per-connection ceilings
128// negotiated exactly once by the daemonmux physical preamble: the receive
129// envelope ceiling (1 MiB..16 MiB), the stream chunk ceiling (1..64 KiB), the
130// concurrent stream ceiling (1..128), and the aggregate buffered-bytes
131// ceiling (MinMuxAggregateBytes..64 MiB). The preamble adopts the
132// element-wise minima of both peers' advertisements as its effective value.
133//
134// The engine and scheduler constructors take a MuxCeilings value and copy it,
135// so the ceilings an instance enforces are fixed for its lifetime: a caller
136// that mutates its own value after construction never changes live
137// enforcement. The zero value is invalid and is refused by Validate.
138type MuxCeilings struct {
139 // MaxReceiveEnvelopeBytes is the largest serialized application envelope
140 // either peer accepts, 1 MiB..16 MiB.
141 MaxReceiveEnvelopeBytes uint64
142 // StreamChunkLimit is the largest opaque stream chunk, 1..64 KiB.
143 StreamChunkLimit uint64
144 // MaxStreams is the largest number of pending plus live streams, 1..128:
145 // opening, open, or closing (a closing stream keeps its slot until its
146 // preserved inbound data is drained).
147 MaxStreams uint64
148 // MaxAggregateBytes is the largest number of buffered bytes across every
149 // live stream, MinMuxAggregateBytes..64 MiB. It is a per-direction budget:
150 // the inbound engine and the outbound scheduler each enforce it
151 // independently, so the two buffered directions never share one
152 // accounting. The floor admits one maximum chunk plus envelope overhead.
153 MaxAggregateBytes uint64
154}
155
156// Validate refuses a zero, below-floor, or above-ceiling advertisement.
157// Envelope bounds are the 1 MiB..16 MiB window, chunk bounds are 1..64 KiB,
158// stream counts are 1..128, and aggregate bounds are
159// MinMuxAggregateBytes..64 MiB (one maximum chunk plus envelope overhead at
160// the floor).
161func (c MuxCeilings) Validate() error {
162 if c.MaxReceiveEnvelopeBytes < MinMuxEnvelopeBytes ||
163 c.MaxReceiveEnvelopeBytes > MaxMuxEnvelopeBytes {
164 return ErrInvalidCeilings
165 }
166 if c.StreamChunkLimit < MinMuxChunkBytes ||
167 c.StreamChunkLimit > MaxMuxChunkBytes {
168 return ErrInvalidCeilings
169 }
170 if c.MaxStreams < MinMuxStreams || c.MaxStreams > MaxMuxStreams {
171 return ErrInvalidCeilings
172 }
173 if c.MaxAggregateBytes < MinMuxAggregateBytes ||
174 c.MaxAggregateBytes > MaxMuxAggregateBytes {
175 return ErrInvalidCeilings
176 }
177 return nil
178}
179
180// DefaultMuxCeilings advertises the local daemonmux receive policy: the full
181// absolute envelope ceiling, the maximum chunk, the maximum stream count,
182// and the maximum aggregate buffered bytes.
183func DefaultMuxCeilings() MuxCeilings {
184 return MuxCeilings{
185 MaxReceiveEnvelopeBytes: MaxMuxEnvelopeBytes,
186 StreamChunkLimit: MaxMuxChunkBytes,
187 MaxStreams: MaxMuxStreams,
188 MaxAggregateBytes: MaxMuxAggregateBytes,
189 }
190}
191
192// EffectiveMuxCeilings takes the element-wise minima of both sides'
193// advertisements: the value the preamble adopts when a peer offers less than
194// the local policy. Each field is independent, so a peer that offers less in
195// one dimension lowers only that dimension. Both inputs are assumed
196// individually valid; the minima of two valid advertisements is valid.
197//
198// The stream ceiling is then lowered, when needed, so every stream can hold a
199// full window of one maximum chunk inside the aggregate budget at once. Both
200// peers apply the same rule, so they agree on the result; the aggregate floor
201// admits at least one stream.
202func EffectiveMuxCeilings(local, remote MuxCeilings) MuxCeilings {
203 effective := MuxCeilings{
204 MaxReceiveEnvelopeBytes: min(local.MaxReceiveEnvelopeBytes, remote.MaxReceiveEnvelopeBytes),
205 StreamChunkLimit: min(local.StreamChunkLimit, remote.StreamChunkLimit),
206 MaxStreams: min(local.MaxStreams, remote.MaxStreams),
207 MaxAggregateBytes: min(local.MaxAggregateBytes, remote.MaxAggregateBytes),
208 }
209 fair := effective.MaxAggregateBytes / chunkCredit(int(effective.StreamChunkLimit))
210 effective.MaxStreams = max(min(effective.MaxStreams, fair), MinMuxStreams)
211 return effective
212}
213
214// checkMuxCeilings refuses an invalid advertisement as a preamble refusal so
215// the preamble keeps its single ErrPreambleRejected sentinel.
216func checkMuxCeilings(ceilings MuxCeilings) error {
217 if err := ceilings.Validate(); err != nil {
218 return ErrPreambleRejected
219 }
220 return nil
221}
222
223// checkMuxEnvelopeCeiling rejects one serialized daemonmux application
224// envelope above either bound: the negotiated receive ceiling or the fixed
225// 1 MiB application ceiling. Every daemonmux envelope is control-class, so
226// the application ceiling always applies in full.
227func checkMuxEnvelopeCeiling(payload []byte, ceiling uint64) error {
228 if uint64(len(payload)) > min(ceiling, MaxMuxApplicationBytes) {
229 return wire.ErrScanLength
230 }
231 return nil
232}
233
234// CheckPreambleSize refuses one serialized daemonmux preamble above the
235// 4 KiB bound before scanning or allocation. Callers apply it to an encoded
236// preamble before sending and to a received payload before decode; the
237// session and broker contracts apply the identical wire.PreambleLimit.
238func CheckPreambleSize(payload []byte) error {
239 if len(payload) > MuxPreambleLimit {
240 return ErrPreambleRejected
241 }
242 return nil
243}