Module: PumaPlus::Wire

Defined in:
lib/puma_plus/wire.rb

Overview

The puma-plus frame protocol, Ruby side.

Mirror of internal/wire/wire.go. See docs/PROTOCOL.md for the normative spec; internal/wire/golden_test.go enforces that the two agree byte for byte.

Deliberately pure Ruby with no dependencies: String#pack/unpack and byteslice are enough for a length-prefixed framing over a flat string->string map, and every dependency avoided is one the gem does not have to install.

Defined Under Namespace

Classes: Closed, PayloadTooLarge, Truncated

Constant Summary collapse

HELLO =

Frame types, data connections.

0x01
REQUEST = 
0x02
BODY_CHUNK = 
0x03
BODY_END = 
0x04
RESPONSE = 
0x05
RESP_CHUNK = 
0x06
RESP_END = 
0x07
HIJACK = 
0x08
GOAWAY = 
0x09
WS_SEND =

Frame types, WebSocket control connections (role "ws"). Outbound commands from Ruby, issuable from any thread at any time rather than only while a message is being handled.

0x50
WS_PUBLISH = 
0x51
WS_SUBSCRIBE = 
0x52
WS_UNSUBSCRIBE = 
0x53
WS_CLOSE = 
0x54
WS_QUERY = 
0x55
WS_QUERY_REPLY = 
0x56
WS_DATAGRAM =

WebTransport-only. Separate frames rather than a flag: best-effort delivery is a different contract from reliable delivery.

0x57
WS_PUBLISH_DATAGRAM = 
0x58
WS_OPEN_STREAM = 
0x59
WORKER_STATUS =

Frame types, control connections.

0x40
SET_SLOTS = 
0x41
QUIESCE = 
0x42
SHUTDOWN = 
0x43
PING = 
0x44
PONG = 
0x45
BODY_NONE =

body_mode values in the REQUEST fixed meta.

0
BODY_INLINE = 
1
BODY_STREAM = 
2
HTTP_10 =

http_version values in the REQUEST fixed meta.

10
HTTP_11 = 
11
HTTP_2 = 
20
HTTP_3 = 
30
VERSION = 
"1"
HEADER_SIZE = 
8
MAX_PAYLOAD = 
16 << 20
CHUNK_TARGET = 
64 << 10
INLINE_BODY_MAX = 
64 << 10
REQUEST_META_SIZE = 
26
RESP_END_META_SIZE = 
32
TYPE_NAMES = 
{
  HELLO => "HELLO", REQUEST => "REQUEST", BODY_CHUNK => "BODY_CHUNK",
  BODY_END => "BODY_END", RESPONSE => "RESPONSE", RESP_CHUNK => "RESP_CHUNK",
  RESP_END => "RESP_END", HIJACK => "HIJACK", GOAWAY => "GOAWAY",
  WORKER_STATUS => "WORKER_STATUS", SET_SLOTS => "SET_SLOTS",
  QUIESCE => "QUIESCE", SHUTDOWN => "SHUTDOWN", PING => "PING", PONG => "PONG",
  WS_SEND => "WS_SEND", WS_PUBLISH => "WS_PUBLISH", WS_SUBSCRIBE => "WS_SUBSCRIBE",
  WS_UNSUBSCRIBE => "WS_UNSUBSCRIBE", WS_CLOSE => "WS_CLOSE",
  WS_QUERY => "WS_QUERY", WS_QUERY_REPLY => "WS_QUERY_REPLY",
  WS_DATAGRAM => "WS_DATAGRAM", WS_PUBLISH_DATAGRAM => "WS_PUBLISH_DATAGRAM",
  WS_OPEN_STREAM => "WS_OPEN_STREAM"
}.freeze
BINARY = 
Encoding::BINARY

Class Method Summary collapse

Class Method Details

.decode_kv(buf, offset = 0) ⇒ Object

Decode a kv blob from the front of buf starting at offset. Returns [pairs, next_offset]. Values are byteslices of buf, not copies -- on the request hot path that aliasing is the point, since the whole env becomes byteslices of a single payload string.

Raises:



126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
 
# File 'lib/puma_plus/wire.rb', line 126

def decode_kv(buf, offset = 0)
  raise Truncated, "kv count" if buf.bytesize - offset < 4

  count = buf.unpack1("N", offset: offset)
  offset += 4

  # A count larger than the remaining bytes can never be satisfied (the
  # smallest possible pair is 6 bytes). Checking up front stops a corrupt
  # length from driving a huge allocation.
  raise Truncated, "kv count #{count} exceeds remaining bytes" if count > buf.bytesize - offset

  pairs = Array.new(count)
  count.times do |i|
    raise Truncated, "kv key length" if buf.bytesize - offset < 2
    kl = buf.unpack1("n", offset: offset)
    offset += 2
    raise Truncated, "kv key" if buf.bytesize - offset < kl
    key = buf.byteslice(offset, kl)
    offset += kl

    raise Truncated, "kv value length" if buf.bytesize - offset < 4
    vl = buf.unpack1("N", offset: offset)
    offset += 4
    raise Truncated, "kv value" if buf.bytesize - offset < vl
    val = buf.byteslice(offset, vl)
    offset += vl

    pairs[i] = [key, val]
  end
  [pairs, offset]
end

.decode_request_meta(buf, offset = 0) ⇒ Object

Decode the fixed 26-byte REQUEST meta from buf at offset. Returns [meta_hash, next_offset].

Raises:



169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
 
# File 'lib/puma_plus/wire.rb', line 169

def decode_request_meta(buf, offset = 0)
  raise Truncated, "request meta" if buf.bytesize - offset < REQUEST_META_SIZE

  queue_ns, runnable_wall_us, body_wait_ns = buf.unpack("Q>3", offset: offset)
  http_version = buf.getbyte(offset + 24)
  body_mode    = buf.getbyte(offset + 25)

  [{
    queue_ns: queue_ns,
    runnable_wall_us: runnable_wall_us,
    body_wait_ns: body_wait_ns,
    http_version: http_version,
    body_mode: body_mode
  }, offset + REQUEST_META_SIZE]
end

.decode_resp_end_meta(buf, offset = 0) ⇒ Object

Decode the fixed 24-byte RESP_END meta. Returns [meta_hash, next_offset].

Raises:



192
193
194
195
196
197
198
199
 
# File 'lib/puma_plus/wire.rb', line 192

def decode_resp_end_meta(buf, offset = 0)
  raise Truncated, "resp_end meta" if buf.bytesize - offset < RESP_END_META_SIZE

  service_ns, cpu_ns, body_read_ns, gvl_wait_ns = buf.unpack("Q>4", offset: offset)
  [{ service_ns: service_ns, cpu_ns: cpu_ns, body_read_ns: body_read_ns,
     gvl_wait_ns: gvl_wait_ns },
   offset + RESP_END_META_SIZE]
end

.encode_kv(pairs) ⇒ Object

Encode a kv blob: u32 count, then count x (u16 klen, k, u32 vlen, v).

pairs is an array of [key, value] rather than a Hash because duplicate keys are meaningful in header blobs, where they encode Rack 3 array-valued headers in wire order.



112
113
114
115
116
117
118
119
120
 
# File 'lib/puma_plus/wire.rb', line 112

def encode_kv(pairs)
  out = [pairs.size].pack("N")
  pairs.each do |k, v|
    k = k.to_s.b
    v = v.to_s.b
    out << [k.bytesize].pack("n") << k << [v.bytesize].pack("N") << v
  end
  out
end

.encode_request_meta(queue_ns:, runnable_wall_us:, body_wait_ns:, http_version:, body_mode:) ⇒ Object

Encode the fixed 26-byte REQUEST meta.



159
160
161
162
163
164
165
 
# File 'lib/puma_plus/wire.rb', line 159

def encode_request_meta(queue_ns:, runnable_wall_us:, body_wait_ns:,
                        http_version:, body_mode:)
  # Ruby has no native u64 pack directive that is endian-explicit in older
  # syntax, so split each into two u32s. Q> exists and is used here.
  [queue_ns, runnable_wall_us, body_wait_ns].pack("Q>3") <<
    [http_version, body_mode].pack("C2")
end

.encode_resp_end_meta(service_ns:, cpu_ns:, body_read_ns:, gvl_wait_ns: 0) ⇒ Object

Encode the fixed 24-byte RESP_END meta. Piggybacking per-request timing on the response costs no extra frame and no extra syscall.



187
188
189
 
# File 'lib/puma_plus/wire.rb', line 187

def encode_resp_end_meta(service_ns:, cpu_ns:, body_read_ns:, gvl_wait_ns: 0)
  [service_ns, cpu_ns, body_read_ns, gvl_wait_ns].pack("Q>4")
end

.frame(type, payload = "") ⇒ Object

Encode a frame: u8 type, u24 reserved (zero), u32 length, payload. Returns a single string so the caller emits one write per frame; frames must never appear interleaved on the wire.



103
104
105
 
# File 'lib/puma_plus/wire.rb', line 103

def frame(type, payload = "")
  [type, 0, 0, 0, payload.bytesize].pack("C4N") << payload.b
end

.read_exactly(io, n) ⇒ Object

Read exactly n bytes. Returns nil on clean EOF before any byte was read, which read_frame maps to Closed; a partial read raises Truncated.

Raises:



228
229
230
231
232
233
234
 
# File 'lib/puma_plus/wire.rb', line 228

def read_exactly(io, n)
  buf = io.read(n)
  return nil if buf.nil?
  raise Truncated, "wanted #{n} bytes, got #{buf.bytesize}" if buf.bytesize < n

  buf
end

.read_frame(io) ⇒ Object

Read one frame from io. Returns [type, payload].

Raises Closed on a clean EOF between frames, Truncated on EOF mid-frame -- the distinction matters because the first is normal shutdown and the second means the peer died holding our request.

Raises:



206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
 
# File 'lib/puma_plus/wire.rb', line 206

def read_frame(io)
  hdr = read_exactly(io, HEADER_SIZE)
  raise Closed if hdr.nil?

  type = hdr.getbyte(0)
  # hdr[1..3] is the reserved u24. Per spec we ignore its value rather than
  # rejecting non-zero, so it can later become a stream id.
  len = hdr.unpack1("N", offset: 4)

  if len > MAX_PAYLOAD
    raise PayloadTooLarge, "#{type_name(type)} announced #{len} bytes"
  end
  return [type, ""] if len.zero?

  payload = read_exactly(io, len)
  raise Truncated, "#{type_name(type)} body" if payload.nil?

  [type, payload]
end

.type_name(type) ⇒ Object 



96
97
98
 
# File 'lib/puma_plus/wire.rb', line 96

def type_name(type)
  TYPE_NAMES[type] || format("UNKNOWN(0x%02x)", type)
end