1// Logical connections over one physical Pump.
2//
3// A LogicalConnection carries complete raw session envelopes across the broker,
4// without decoding them. Typed preview and observation are adapted by their
5// consumers, never by this mux connection.
6//
7// The carriage between the two is a private, ordered, bounded byte stream
8// (muxStreamPipe). Its read side pulls already-queued chunks from the pump
9// engine with Take, waking on the pump's per-stream Watch; the pump reader is
10// therefore never blocked by a slow or stalled consumer, and a consumer that
11// never reads lets the engine reset only its own stream. Its write side splits
12// each outbound byte slice into chunks at or below the negotiated
13// StreamChunkLimit and hands them to Pump.Send as Data frames, so streamframe
14// sees one ordered byte stream and applies the 4-byte session framing on top.
15//
16// The open side maps a ports.BrokerOpenStreamRequest to one mux Open and its
17// StreamRef, admits the stream locally, sends the Open, and waits under the
18// caller's context for the daemon's Opened or Refused. An independent Close
19// sends one mux Close and settles orderly; a malformed inner session frame
20// resets exactly that stream (one mux Reset) and leaves every sibling and the
21// physical connection untouched. Done, Err, and a physical-loss
22// ports.BrokerStreamLost are published once and stay stable afterwards.
23package daemonmux
24
25import (
26 "context"
27 "errors"
28 "io"
29 "sync"
30
31 "github.com/bnema/vev/internal/adapters/streamframe"
32 "github.com/bnema/vev/internal/domain"
33 "github.com/bnema/vev/internal/ports"
34 "github.com/bnema/vev/internal/protocol/wire"
35)
36
37// Logical connection sentinels. They are typed so a caller can classify a
38// rejection without matching on message text.
39var (
40 // ErrLogicalConfig reports an invalid logical-connection configuration: a
41 // nil or server-side pump, or an unencodable open request.
42 ErrLogicalConfig = errors.New("daemonmux: invalid logical connection configuration")
43 // ErrLogicalClosed reports session work presented to a logical connection
44 // that already reached its terminal outcome.
45 ErrLogicalClosed = errors.New("daemonmux: logical stream is closed")
46)
47
48// malformedResetText is the bounded, presentation-safe refusal text carried by
49// the stream-local Reset a logical connection sends when the inner session
50// stream is malformed. The diagnostic cause stays local in the terminal error.
51const malformedResetText = "malformed session stream"
52
53// LogicalConnector opens typed logical streams over one client-side Pump. It
54// owns the mux-local physical stream identity allocator: physical IDs are
55// strictly increasing and never reused, as the engine requires. One connector
56// is bound to exactly one physical connection. Its Open is safe for concurrent
57// use; ID allocation and local admission are serialized so engine admission
58// always sees strictly increasing IDs, while the Opened/Refused wait runs
59// outside that lock.
60type LogicalConnector struct {
61 pump *Pump
62
63 openMu sync.Mutex
64 next uint64
65}
66
67// NewLogicalConnector returns a connector over one broker-side pump: the pump
68// must decode daemon-to-broker frames (Inbound == DirectionServer) and emit
69// broker-to-daemon frames. A nil, engine-less, or server-side pump is refused.
70func NewLogicalConnector(pump *Pump) (*LogicalConnector, error) {
71 if pump == nil || pump.Engine() == nil || pump.Local() != DirectionClient {
72 return nil, ErrLogicalConfig
73 }
74 return &LogicalConnector{pump: pump}, nil
75}
76
77// Pump returns the physical pump this connector opens streams over.
78func (c *LogicalConnector) Pump() *Pump { return c.pump }
79
80// Open maps request to one mux Open and StreamRef, admits the stream locally,
81// sends the Open, and waits under ctx for the daemon's Opened or Refused. A
82// locally admitted refusal, a stream-reset, or a physical loss during the wait
83// is returned as a typed error and the local reservation is released. A
84// cancelled context releases the local reservation and returns ctx.Err().
85func (c *LogicalConnector) Open(ctx context.Context, request ports.BrokerOpenStreamRequest) (*LogicalConnection, error) {
86 if c == nil || c.pump == nil {
87 return nil, ErrLogicalConfig
88 }
89 if ctx == nil {
90 ctx = context.Background()
91 }
92 physical, err := c.reserve(request)
93 if err != nil {
94 return nil, err
95 }
96 status, err := c.awaitOpened(ctx, physical)
97 if err != nil {
98 c.pump.ReleaseWatch(physical)
99 _ = c.abortReserved(physical)
100 return nil, err
101 }
102 return newLogicalConnection(c.pump, status.Ref, status.Done, c.pump.cause(physical)), nil
103}
104
105// awaitOpened blocks until the admitted stream is confirmed open, refused, or
106// settles. It watches the pump's per-stream channel so it never polls, and it
107// also honors ctx and a physical terminal outcome.
108func (c *LogicalConnector) awaitOpened(ctx context.Context, physical PhysicalStreamID) (StreamStatus, error) {
109 engine := c.pump.Engine()
110 for {
111 watch := c.pump.Watch(physical)
112 status, ok := engine.Status(physical)
113 if !ok {
114 return StreamStatus{}, openFailure(ErrInvalidStream, c.pump.Err())
115 }
116 switch status.State {
117 case StreamOpen:
118 return status, nil
119 case StreamTerminal:
120 return StreamStatus{}, refusedOrLost(status, c.pump.Err())
121 }
122 select {
123 case <-watch:
124 case <-ctx.Done():
125 return StreamStatus{}, ctx.Err()
126 case <-c.pump.Done():
127 status, ok := engine.Status(physical)
128 if !ok {
129 return StreamStatus{}, openFailure(c.pump.Err(), c.pump.Err())
130 }
131 return StreamStatus{}, refusedOrLost(status, c.pump.Err())
132 }
133 }
134}
135
136// abortReserved releases one locally admitted reservation that never became a
137// live connection: it settles the engine record and best-effort signals the
138// peer with a Reset, without touching the physical connection or a sibling.
139func (c *LogicalConnector) abortReserved(physical PhysicalStreamID) error {
140 reset := Reset{Physical: physical}
141 _, _ = c.pump.Engine().Reset(reset)
142 _ = c.pump.Send(reset)
143 return nil
144}
145
146// openMessage maps one validated request to its mux Open and StreamRef.
147func openMessage(physical PhysicalStreamID, request ports.BrokerOpenStreamRequest) Open {
148 return Open{
149 Ref: StreamRef{
150 Physical: physical,
151 Epoch: request.Epoch,
152 Connection: request.Connection,
153 Client: request.Stream,
154 },
155 Purpose: request.Purpose,
156 Admission: request.Admission,
157 Name: request.Name,
158 Local: request.Local,
159 Endpoint: request.Endpoint,
160 Registration: request.Registration,
161 Target: request.Target,
162 Env: append([]string(nil), request.Env...),
163 Policy: request.Policy,
164 // The daemon-start authorization travels with the mux Open, so the
165 // daemon-side transport receives exactly the authority the broker
166 // resolved and can never widen it into a spawn.
167 StartMode: request.StartMode,
168 }
169}
170
171// openAdmissionError classifies one refused local admission as a typed
172// admission error so the caller can retry with a fresh identity.
173func openAdmissionError(err error) error {
174 switch {
175 case errors.Is(err, ErrTooManyStreams):
176 return ports.BrokerAdmissionLimit
177 case errors.Is(err, ErrPhysicalClosed):
178 return ports.BrokerAdmissionClosed
179 case errors.Is(err, ErrStreamIDReused):
180 return ports.BrokerAdmissionStale
181 default:
182 return ports.BrokerAdmissionInvalid
183 }
184}
185
186// refusedOrLost classifies one terminal status observed while opening: an
187// explicit refusal, or a stream reset / physical loss.
188func refusedOrLost(status StreamStatus, physicalErr error) error {
189 if status.Refused {
190 if status.Refusal.AdmissionCode != 0 {
191 switch status.Refusal.AdmissionCode {
192 case 1:
193 return ports.BrokerAdmissionLimit
194 case 2:
195 return ports.BrokerAdmissionClosed
196 case 3:
197 return ports.BrokerAdmissionStale
198 case 4:
199 return ports.BrokerAdmissionInvalid
200 }
201 }
202 return ports.BrokerError{Code: status.Refusal.Code, Text: status.Refusal.Text}
203 }
204 if physicalErr != nil {
205 return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Cause: physicalErr}
206 }
207 return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Cause: status.Err}
208}
209
210// openFailure maps one non-status transport failure on open.
211func openFailure(err error, physicalErr error) error {
212 if physicalErr != nil {
213 return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Cause: physicalErr}
214 }
215 return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Cause: err}
216}
217
218// discardQueuedInbound drains every inbound chunk the engine already accepted
219// for one physical stream, releasing its bytes from the per-stream and
220// aggregate budgets. A locally closing stream calls it before the engine's
221// orderly Close, because a local Close stops this side's reader: nothing else
222// would ever drain a preserved queue, and the engine only reaches its terminal
223// state - releasing the admission slot and the queued bytes - once that queue
224// is empty. It is a no-op for an unknown, empty, or already terminal stream.
225func discardQueuedInbound(pump *Pump, physical PhysicalStreamID) {
226 for {
227 if _, ok := pump.Take(physical); !ok {
228 return
229 }
230 }
231}
232
233// LogicalConnection is one client-side logical stream over a physical Pump.
234// It implements the raw envelope port; Close, Done and Err manage its lifetime.
235type LogicalConnection struct {
236 pump *Pump
237 ref StreamRef
238 pipe *muxStreamPipe
239 raw *muxTransport
240 stream <-chan struct{}
241 // cause is the stream's retained publish-once terminal authority, captured
242 // at open. It outlives the engine's bounded record, so a settled stream
243 // keeps its exact cause even after the record is evicted.
244 cause *terminalState
245
246 terminal *terminalState
247 watchDone chan struct{}
248 closeOnce sync.Once
249 closeErr error
250}
251
252var _ ports.BrokerEnvelopeStream = (*LogicalConnection)(nil)
253
254// newLogicalConnection builds the framed carriage over one confirmed open
255// stream and starts the terminal watcher. It never blocks. cause is the
256// stream's retained terminal authority, captured at open so the connection
257// reports its exact outcome independent of the engine's bounded record.
258func newLogicalConnection(pump *Pump, ref StreamRef, streamDone <-chan struct{}, cause *terminalState) *LogicalConnection {
259 limit := pump.Ceilings().StreamChunkLimit
260 pipe := newMuxStreamPipe(pump, ref.Physical, limit, streamDone)
261 raw := newMuxTransport(pipe)
262 connection := &LogicalConnection{
263 pump: pump,
264 ref: ref,
265 pipe: pipe,
266 raw: raw,
267 stream: streamDone,
268 cause: cause,
269 terminal: newTerminalState(),
270 watchDone: make(chan struct{}),
271 }
272 go connection.watch()
273 return connection
274}
275
276// Ref returns the complete stream reference recorded at admission.
277func (c *LogicalConnection) Ref() StreamRef { return c.ref }
278
279// Done returns the logical terminal channel independent of reads. It closes
280// exactly once, before Err becomes stable.
281func (c *LogicalConnection) Done() <-chan struct{} { return c.terminal.Done() }
282
283// Err returns the stable terminal cause: nil for an orderly local or peer
284// Close, a ports.BrokerStreamLost for a peer Reset, a local malformed-inner
285// reset, or a physical loss.
286func (c *LogicalConnection) Err() error { return c.terminal.Err() }
287
288// FailureKind classifies Err. It is None while open and for an orderly Close.
289func (c *LogicalConnection) FailureKind() domain.RemoteFailureKind { return c.terminal.FailureKind() }
290
291func (c *LogicalConnection) SendEnvelope(payload []byte) error {
292 if c.settled() {
293 return c.terminalError()
294 }
295 if len(payload) > wire.AbsoluteEnvelopeLimit {
296 return streamframe.ErrTooLarge
297 }
298 return c.raw.Send(wire.Envelope{Payload: payload})
299}
300
301func (c *LogicalConnection) RecvEnvelope() ([]byte, error) {
302 envelope, err := c.raw.Recv()
303 if err != nil {
304 if errors.Is(err, io.EOF) {
305 c.syncFromEngine()
306 return nil, c.terminalError()
307 }
308 return nil, c.abortMalformed(err)
309 }
310 return envelope.Payload, nil
311}
312
313// Close independently closes the logical stream: it discards the inbound
314// chunks the engine already accepted, settles this stream locally, sends
315// exactly one mux Close, publishes the orderly nil outcome, and unblocks every
316// read and write. It never touches the physical connection or a sibling
317// stream. Close is idempotent and concurrent-safe; every caller observes the
318// same stored transport Close error.
319func (c *LogicalConnection) Close() error {
320 if c == nil {
321 return nil
322 }
323 c.closeOnce.Do(func() {
324 // Discard what the engine already accepted before closing locally: this
325 // side stops reading as soon as it closes, so draining first is what lets
326 // the engine's orderly Close reach its terminal state and release the
327 // stream's admission slot and aggregate bytes instead of parking the
328 // stream in closing with a queue nothing will ever drain.
329 discardQueuedInbound(c.pump, c.ref.Physical)
330 _, _ = c.pump.Engine().Close(Close{Physical: c.ref.Physical})
331 // A chunk the reader accepted between the drain and the Close parked the
332 // stream in closing instead of terminal; no new chunk is accepted after
333 // the Close, so this drain terminalizes the stream and releases its slot
334 // and bytes.
335 discardQueuedInbound(c.pump, c.ref.Physical)
336 _ = c.pump.Send(Close{Physical: c.ref.Physical})
337 c.terminal.Close()
338 c.closeErr = c.raw.Close()
339 <-c.watchDone
340 c.pump.ReleaseWatch(c.ref.Physical)
341 })
342 return c.closeErr
343}
344
345// watch publishes the logical terminal outcome when the stream settles on its
346// own (peer Close, peer Reset, queue overflow, or physical loss) independently
347// of reads, and unblocks the byte stream. A local Close settles first and makes
348// this a no-op.
349func (c *LogicalConnection) watch() {
350 defer c.pump.ReleaseWatch(c.ref.Physical)
351 defer close(c.watchDone)
352 select {
353 case <-c.stream:
354 // The stream reached its terminal state. Publish the retained cause,
355 // not a fresh engine lookup: the engine record may already be evicted.
356 c.syncFromEngine()
357 case <-c.pump.Done():
358 // A physical terminal outcome wakes every stream. A stream-local cause
359 // recorded before the physical failure stays authoritative.
360 if !c.settleFromCause() {
361 if err := c.pump.Err(); err != nil {
362 c.settleLoss(c.pump.FailureKind(), err)
363 } else {
364 c.terminal.Close()
365 }
366 }
367 case <-c.pipe.Done():
368 return
369 }
370 // Closing the framed transport also closes the byte stream and joins
371 // streamframe's writer, so a self-terminating stream leaves no goroutine
372 // behind.
373 _ = c.raw.Close()
374}
375
376// syncFromEngine settles this connection from the stream's retained terminal
377// authority when it published, and otherwise from the pump's own terminal
378// outcome. It closes the race between a blocked read observing the inner-stream
379// end and the watcher publishing the same outcome, and it never lets a physical
380// failure surface as a clean io.EOF: the retained authority outlives the
381// bounded engine record, so an evicted stream still reports its exact cause.
382func (c *LogicalConnection) syncFromEngine() {
383 if c.settled() {
384 return
385 }
386 if c.settleFromCause() {
387 return
388 }
389 if err := c.pump.Err(); err != nil {
390 c.settleLoss(c.pump.FailureKind(), err)
391 return
392 }
393 if status, ok := c.pump.Engine().Status(c.ref.Physical); ok && status.State == StreamTerminal {
394 c.settleFromStatus(status)
395 return
396 }
397 c.terminal.Close()
398}
399
400// settleFromCause publishes the terminal outcome recorded by the stream's
401// retained publish-once authority. The authority is captured when the stream
402// opens and stays valid for the connection's lifetime even after the engine
403// evicts the bounded record, so a settled stream keeps its exact cause
404// independent of retention: a Reset never degrades into an orderly close. It
405// reports whether the authority had published.
406func (c *LogicalConnection) settleFromCause() bool {
407 cause := c.cause
408 if cause == nil {
409 return false
410 }
411 select {
412 case <-cause.Done():
413 default:
414 return false
415 }
416 if err := cause.Err(); err != nil {
417 c.settleLoss(cause.FailureKind(), err)
418 } else {
419 c.terminal.Close()
420 }
421 return true
422}
423
424// settleFromStatus publishes the terminal outcome recorded by the engine.
425func (c *LogicalConnection) settleFromStatus(status StreamStatus) {
426 switch {
427 case status.Refused:
428 c.terminal.Fail(domain.RemoteFailureInvalidResponse, ports.BrokerError{Code: status.Refusal.Code, Text: status.Refusal.Text})
429 case status.Err == nil:
430 c.terminal.Close()
431 default:
432 c.settleLoss(status.FailureKind, status.Err)
433 }
434}
435
436// abortMalformed resets exactly this stream after a malformed inner session
437// frame: it settles the local engine record, best-effort signals the peer with
438// one Reset, publishes the stream-local failure, unblocks the byte stream, and
439// returns the stable terminal error.
440func (c *LogicalConnection) abortMalformed(cause error) error {
441 reset := Reset{
442 Physical: c.ref.Physical,
443 Error: ErrorDetail{
444 Code: ports.BrokerErrorUnavailable,
445 Text: malformedResetText,
446 FailureKind: domain.RemoteFailureInvalidResponse,
447 },
448 HasError: true,
449 }
450 _, _ = c.pump.Engine().Reset(reset)
451 _ = c.pump.Send(reset)
452 c.settleLoss(domain.RemoteFailureInvalidResponse, cause)
453 _ = c.raw.Close()
454 return c.terminalError()
455}
456
457// settleLoss publishes one stream-local loss carrying the exact
458// epoch/connection/stream scope of this logical connection.
459func (c *LogicalConnection) settleLoss(kind domain.RemoteFailureKind, cause error) {
460 if kind < domain.RemoteFailureTransport || kind > domain.RemoteFailureInvalidResponse {
461 kind = domain.RemoteFailureTransport
462 }
463 c.terminal.Fail(kind, ports.BrokerStreamLost{
464 Connection: c.ref.Connection, Stream: c.ref.Client, Epoch: c.ref.Epoch,
465 Cause: kind, Err: cause,
466 })
467}
468
469// settled reports whether the logical connection already reached its terminal
470// outcome.
471func (c *LogicalConnection) settled() bool {
472 select {
473 case <-c.terminal.Done():
474 return true
475 default:
476 return false
477 }
478}
479
480// terminalError is the stable error of a settled connection: the published
481// cause, or io.EOF for an orderly close.
482func (c *LogicalConnection) terminalError() error {
483 if err := c.terminal.Err(); err != nil {
484 return err
485 }
486 return io.EOF
487}
488
489// muxStreamPipe is the private ordered, bounded byte stream of one logical mux
490// stream. Read pulls already-queued inbound chunks from the pump engine with
491// Take, blocking on the pump's per-stream Watch and ending on the stream's
492// terminal channel or the pump's Done; Write splits an outbound slice at the
493// negotiated chunk ceiling and sends each piece as one Data frame. Close
494// unblocks both directions. Exactly one reader is expected (streamframe
495// serializes Recv), and concurrent writes are serialized by streamframe too.
496type muxStreamPipe struct {
497 pump *Pump
498 id PhysicalStreamID
499 limit uint64
500 streamDone <-chan struct{}
501 done chan struct{}
502 once sync.Once
503 mu sync.Mutex
504 pending []byte
505}
506
507// newMuxStreamPipe builds the ordered byte stream of one mux stream. limit is
508// the negotiated StreamChunkLimit; it must be nonzero.
509func newMuxStreamPipe(pump *Pump, id PhysicalStreamID, limit uint64, streamDone <-chan struct{}) *muxStreamPipe {
510 if limit == 0 {
511 limit = MinMuxChunkBytes
512 }
513 return &muxStreamPipe{pump: pump, id: id, limit: limit, streamDone: streamDone, done: make(chan struct{})}
514}
515
516// done returns the pipe's closed-on-close channel.
517func (s *muxStreamPipe) Done() <-chan struct{} { return s.done }
518
519func (s *muxStreamPipe) closed() bool {
520 select {
521 case <-s.done:
522 return true
523 default:
524 return false
525 }
526}
527
528// Read returns the next ordered inbound bytes. It drains whatever the engine
529// still holds and returns io.EOF once the stream or the physical connection
530// reaches its terminal outcome.
531func (s *muxStreamPipe) Read(p []byte) (int, error) {
532 if len(p) == 0 {
533 return 0, nil
534 }
535 for {
536 s.mu.Lock()
537 if len(s.pending) > 0 {
538 n := copy(p, s.pending)
539 s.pending = s.pending[n:]
540 s.mu.Unlock()
541 return n, nil
542 }
543 s.mu.Unlock()
544 if s.closed() {
545 return 0, io.EOF
546 }
547 watch := s.pump.Watch(s.id)
548 if chunk, ok := s.pump.Take(s.id); ok {
549 s.mu.Lock()
550 s.pending = chunk
551 s.mu.Unlock()
552 continue
553 }
554 if watchSettled(watch) {
555 // The stream is terminal or unknown, so no further inbound chunk
556 // can ever be queued. A live stream's watch may also be closed
557 // already, by a frame that arrived after Take: the select below
558 // then returns at once and the loop takes it.
559 return 0, io.EOF
560 }
561 select {
562 case <-watch:
563 case <-s.done:
564 return 0, io.EOF
565 case <-s.streamDone:
566 return 0, io.EOF
567 case <-s.pump.Done():
568 return 0, io.EOF
569 }
570 }
571}
572
573// Write splits p into ordered chunks at or below the negotiated chunk ceiling
574// and sends each as one Data frame under the stream's flow-control credit: it
575// blocks while the peer has not granted credit, so a slow link slows the
576// writer instead of resetting the stream. It reports the bytes fully sent
577// before an error.
578func (s *muxStreamPipe) Write(p []byte) (int, error) {
579 written := 0
580 for len(p) > 0 {
581 if s.closed() {
582 return written, ErrLogicalClosed
583 }
584 n := len(p)
585 if uint64(n) > s.limit {
586 n = int(s.limit)
587 }
588 if err := s.pump.SendData(s.id, p[:n], s.done); err != nil {
589 if s.closed() || errors.Is(err, errStreamSettled) {
590 return written, ErrLogicalClosed
591 }
592 return written, err
593 }
594 written += n
595 p = p[n:]
596 }
597 return written, nil
598}
599
600// Close unblocks every read and write on the byte stream exactly once.
601func (s *muxStreamPipe) Close() error {
602 s.once.Do(func() { close(s.done) })
603 return nil
604}
605
606// muxTransport adapts the ordered byte stream to wire.Transport: streamframe
607// owns the 4-byte length framing, and sessionwire owns any typed conversion at the consuming endpoint. Closing it closes the byte stream and joins streamframe's writer.
608type muxTransport struct {
609 pipe *muxStreamPipe
610 framer *streamframe.Framer
611}
612
613var _ wire.Transport = (*muxTransport)(nil)
614
615// newMuxTransport frames the byte stream into typed envelopes.
616func newMuxTransport(pipe *muxStreamPipe) *muxTransport {
617 return &muxTransport{pipe: pipe, framer: streamframe.NewFramer(pipe, pipe, pipe.Close)}
618}
619
620func (t *muxTransport) Send(envelope wire.Envelope) error { return t.framer.Send(envelope.Payload) }
621
622func (t *muxTransport) Recv() (wire.Envelope, error) {
623 payload, err := t.framer.Recv()
624 if err != nil {
625 return wire.Envelope{}, err
626 }
627 return wire.Envelope{Payload: payload}, nil
628}
629
630func (t *muxTransport) Close() error { return t.framer.Close() }
631
632// reserve allocates the next strictly increasing mux-local physical stream ID,
633// admits it locally, and enqueues its Open frame under one lock, so both the
634// engine's strictly increasing admission order and the outbound frame order
635// match the allocation order even when Opens run concurrently. The lock is
636// released before the open waits on the round trip, so a slow peer never blocks
637// a sibling reservation.
638func (c *LogicalConnector) reserve(request ports.BrokerOpenStreamRequest) (PhysicalStreamID, error) {
639 c.openMu.Lock()
640 defer c.openMu.Unlock()
641 c.next++
642 physical := PhysicalStreamID(c.next)
643 open := openMessage(physical, request)
644 if err := c.pump.Engine().Open(open); err != nil {
645 return 0, openAdmissionError(err)
646 }
647 if err := c.pump.Send(open); err != nil {
648 _ = c.abortReserved(physical)
649 return 0, openFailure(err, c.pump.Err())
650 }
651 return physical, nil
652}