diff --git a/CHANGELOG.md b/CHANGELOG.md index 778eecb..f1a1510 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,54 @@ the [versioning policy](https://teqpace-services.github.io/isopace/versioning/). ## [Unreleased] +## [1.1.0] - 2026-09-11 + +Network management for a persistent acquirer link. Four gaps found while tracing +switch instability at a downstream acquirer, each of which forced a workaround that +could not fully fix the problem from outside the library. Everything here is +**additive** — an existing configuration behaves exactly as before — and confined to +`mux` and `connector`, which v1.0.0 declares **experimental**. + +### Added + +- **Reply to an unsolicited frame.** `mux.Reply`, `mux.WithUnsolicitedReplier`, + `(*mux.Mux).Send`, `(*connector.Connector).Send`, and + `connector.Config.OnUnsolicited`. Previously `Unsolicited` handed over the frame and + nothing else, and nothing exposed a raw send, so a peer-initiated message could not be + answered at all — `Request` is the wrong tool, since the frame we would send *is* the + response and there is nothing to wait for. This matters on real links: Postilion-family + hosts do not only answer echo tests, they send them, and a host that echo-tests us and + never receives an 0810 concludes the link is dead and closes it. +- **`mux.WithResponseKeyer` / `connector.Config.ResponseKeyer`** — a separate keyer for + INBOUND frames. With one keyer serving both directions no keyer can both match a + response to its request and separate a peer's request from our own; the two are + structurally identical. Network management correlates on the trace number alone (an + echo carries no terminal id), so a peer-originated 0800 on a colliding trace was + delivered as though it were the 0810 answer, and the real answer then arrived + unmatched. With both keyers the caller keys on message type as well as trace: the + request keyer maps an outgoing frame to the key of the response it *expects*, the + response keyer maps an inbound frame by its *own* MTI. +- **`connector.Config.KeepaliveTimeout`** — bounds one keepalive independently of + `Timeout` (0 falls back to `Timeout`, the previous behaviour). Worth stating precisely, + because the obvious diagnosis is wrong: a peer that dies cleanly (EOF/RST) fails the + mux from the read loop, which unblocks the pending keepalive at once. The gap is the + **half-open** peer — TCP established, application silent, no FIN — where nothing closes + the mux and the echo waits out a timeout sized for a financial round-trip, while the + connector goes on reporting `Connected()` and accepting requests onto a socket that + will never answer. +- **`connector.Config.OnReady`** — runs after the connector is marked up, in its own + goroutine, on each (re)connect. `OnConnect` runs *before* the mux is published, which + is right for sign-on but leaves nowhere for a ceremony that must go through the + connector's own request path (a working-key exchange, a parameter download): from + `OnConnect` it fails `ErrNotConnected`. Unlike `OnConnect` it cannot drop the link — + its failure is the caller's to handle. + +### Changed + +- `(*connector.Connector).serve` runs the keepalive off the supervise loop. This does + **not** shrink the half-open detection window (see `KeepaliveTimeout` above) but it + stops an outstanding keepalive from holding `Stop` hostage. + ## [1.0.0] - 2026-06-17 The first **stable** release. With the release-candidate soak complete, the @@ -276,7 +324,8 @@ tested under `go test -race`. authenticates in constant time; expose it only behind appropriate network controls. -[Unreleased]: https://github.com/teqpace-services/isopace/compare/v1.0.0...HEAD +[Unreleased]: https://github.com/teqpace-services/isopace/compare/v1.1.0...HEAD +[1.1.0]: https://github.com/teqpace-services/isopace/compare/v1.0.0...v1.1.0 [1.0.0]: https://github.com/teqpace-services/isopace/compare/v1.0.0-rc.1...v1.0.0 [1.0.0-rc.1]: https://github.com/teqpace-services/isopace/compare/v0.3.0...v1.0.0-rc.1 [0.3.0]: https://github.com/teqpace-services/isopace/compare/v0.2.0...v0.3.0 diff --git a/connector/connector.go b/connector/connector.go index 4c3e5a9..7c6070b 100644 --- a/connector/connector.go +++ b/connector/connector.go @@ -55,8 +55,14 @@ type Config struct { Network string Addr string // Keyer correlates a response to its in-flight request (typically STAN + - // terminal, via mux.FieldKeyer). + // terminal, via mux.FieldKeyer). With ResponseKeyer unset it keys both + // directions; with it set, this keys OUTGOING requests only. Keyer mux.Keyer + // ResponseKeyer keys INBOUND frames, when they must be keyed differently from + // outgoing ones — see [mux.WithResponseKeyer]. Set it (together with a Keyer that + // maps a request to the key of the response it expects) to stop a peer-originated + // request being mistaken for the answer to one of ours. Optional. + ResponseKeyer mux.Keyer // LinkOptions configure the underlying link: framer (e.g. a Postilion 2-byte // length prefix), TLS, MAC filters, TCP keep-alive. LinkOptions []link.Option @@ -68,15 +74,45 @@ type Config struct { // the connector is marked up — the place for sign-on / session-key exchange. A // non-nil error drops the link and triggers another reconnect. Optional. OnConnect func(ctx context.Context, m *mux.Mux) error + // OnReady runs after the connector is marked UP, in its own goroutine, on each + // (re)connect. It is where a ceremony that must go through the connector's own + // request path belongs — a working-key exchange, a parameter download — because + // OnConnect runs BEFORE the mux is published and a [Connector.Request] from there + // fails ErrNotConnected. Its failure is the caller's to handle: unlike OnConnect, + // it cannot drop the link. Optional. + OnReady func(ctx context.Context, c *Connector) // Keepalive runs on the live mux every KeepaliveInterval — typically an echo // (0800) exchange. A non-nil error drops the link and forces a reconnect, so a // dead-but-open peer is detected without waiting for a real request. Optional. Keepalive func(ctx context.Context, m *mux.Mux) error KeepaliveInterval time.Duration + // KeepaliveTimeout bounds ONE keepalive, independently of Timeout. When it expires + // the link is dropped and reconnected. + // + // Without it a keepalive inherits the request Timeout, which is sized for a + // financial round-trip and is far too generous for an echo. That matters for the + // one failure mode a keepalive exists to catch: a HALF-OPEN peer — TCP still + // established, application silent, no FIN — where nothing ever closes the mux, so + // the echo simply waits. Until it gives up the connector goes on reporting + // Connected() and accepting requests onto a socket that will never answer, for + // KeepaliveInterval + Timeout. Bounding the echo to a few seconds is what shrinks + // that window; a peer that dies cleanly (EOF/RST) is noticed immediately either way, + // because the read loop fails the mux and unblocks the pending request. + // + // 0 falls back to Timeout (the previous behaviour). + KeepaliveTimeout time.Duration // Unsolicited handles frames the peer sends that match no in-flight request — // server-initiated advices, reversals, or sign-off. It runs on the read loop, // so it must not block (hand off to a queue). Optional. + // + // Use OnUnsolicited instead when the frame must be ANSWERED. Unsolicited func(frame []byte) + // OnUnsolicited is Unsolicited with the ability to reply over the same link, and + // takes precedence over it. A peer that echo-tests us (0800) and never receives an + // 0810 concludes the link is dead and closes it, which is indistinguishable from a + // flapping network; answering inline is a single write and is the intended use. + // Optional. + OnUnsolicited func(frame []byte, reply mux.Reply) // Log receives lifecycle events (default slog.Default). Log *slog.Logger } @@ -157,6 +193,17 @@ func (c *Connector) Request(ctx context.Context, req []byte) ([]byte, error) { return m.Request(ctx, req) } +// Send writes req over the live connection WITHOUT waiting for a response — for +// answering a peer-initiated frame, or pushing an advice the peer will not reply to. +// It returns ErrNotConnected if the link is currently down. +func (c *Connector) Send(req []byte) error { + m := c.mux.Load() + if m == nil { + return ErrNotConnected + } + return m.Send(req) +} + // Connected reports whether the link is currently up. func (c *Connector) Connected() bool { return c.mux.Load() != nil } @@ -188,6 +235,12 @@ func (c *Connector) supervise(ctx context.Context) { slog.String("connector", c.cfg.Name), slog.String("addr", c.cfg.Addr)) up := time.Now() + // The link is published, so OnReady may use Request. Its own goroutine: it can + // take as long as its ceremony needs without stalling the keep-alive. + if c.cfg.OnReady != nil { + go c.cfg.OnReady(ctx, c) + } + c.serve(ctx, m) // blocks until the link dies or ctx is cancelled cause := m.Err() c.mux.CompareAndSwap(m, nil) @@ -222,7 +275,13 @@ func (c *Connector) connect(ctx context.Context) (*mux.Mux, error) { if c.cfg.Timeout > 0 { opts = append(opts, mux.WithTimeout(c.cfg.Timeout)) } - if c.cfg.Unsolicited != nil { + if c.cfg.ResponseKeyer != nil { + opts = append(opts, mux.WithResponseKeyer(c.cfg.ResponseKeyer)) + } + switch { + case c.cfg.OnUnsolicited != nil: + opts = append(opts, mux.WithUnsolicitedReplier(c.cfg.OnUnsolicited)) + case c.cfg.Unsolicited != nil: opts = append(opts, mux.WithUnsolicitedHandler(c.cfg.Unsolicited)) } m := mux.New(l, c.cfg.Keyer, opts...) @@ -237,6 +296,14 @@ func (c *Connector) connect(ctx context.Context) (*mux.Mux, error) { // serve runs the keepalive ticker (if any) and blocks until the link dies or ctx // is cancelled. +// +// The keepalive runs in its own goroutine, NOT inline. Run inline it owns the loop +// for as long as the peer takes to answer, and during that time this select cannot +// observe m.Done() — so a link that dies mid-keepalive goes unnoticed, while the +// connector still reports Connected() and accepts requests onto the dead socket. A +// dead-but-open peer (a NAT idle-out, a wedged host: TCP established, application +// silent) therefore stayed "up" for a whole KeepaliveInterval + Timeout. Off the loop, +// the link's death is seen the moment it happens whatever the keepalive is doing. func (c *Connector) serve(ctx context.Context, m *mux.Mux) { var tick <-chan time.Time if c.cfg.Keepalive != nil && c.cfg.KeepaliveInterval > 0 { @@ -244,18 +311,42 @@ func (c *Connector) serve(ctx context.Context, m *mux.Mux) { defer t.Stop() tick = t.C } + // Non-nil while a keepalive is outstanding, so at most one is ever in flight. + var inflight chan error for { select { case <-ctx.Done(): return case <-m.Done(): return - case <-tick: - if err := c.cfg.Keepalive(ctx, m); err != nil { + case err := <-inflight: + inflight = nil + if err != nil { c.log.LogAttrs(ctx, slog.LevelWarn, "connector keepalive failed", slog.String("connector", c.cfg.Name), slog.String("error", err.Error())) return // force a reconnect } + case <-tick: + if inflight != nil { + // The previous keepalive has not come back yet, so the peer is already + // late. Skip this tick rather than stacking a second one: the outstanding + // request carries its own deadline and will report the failure itself. + c.log.LogAttrs(ctx, slog.LevelDebug, "connector keepalive still outstanding, skipping tick", + slog.String("connector", c.cfg.Name)) + continue + } + // Buffered, so the goroutine never blocks writing its result even if this + // loop has already returned and nothing is left to read it. + inflight = make(chan error, 1) + go func(done chan<- error) { + kctx := ctx + if c.cfg.KeepaliveTimeout > 0 { + var cancel context.CancelFunc + kctx, cancel = context.WithTimeout(ctx, c.cfg.KeepaliveTimeout) + defer cancel() + } + done <- c.cfg.Keepalive(kctx, m) + }(inflight) } } } diff --git a/connector/netmgmt_test.go b/connector/netmgmt_test.go new file mode 100644 index 0000000..da5a980 --- /dev/null +++ b/connector/netmgmt_test.go @@ -0,0 +1,412 @@ +package connector + +import ( + "context" + "errors" + "fmt" + "net" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/teqpace-services/isopace/link" + "github.com/teqpace-services/isopace/mux" +) + +// --- a tiny line-framed peer, enough to exercise the connector's hooks ------- + +// testPeer speaks newline-free 2-byte length frames of the form ":". +type testPeer struct { + ln net.Listener + mu sync.Mutex + silent bool // stop answering, but hold the socket open (a dead-but-open peer) + conn net.Conn // the accepted connection; closing the LISTENER does not close this + inbound []string // every frame we received + send chan string +} + +func newTestPeer(t *testing.T, answer func(frame string) (string, bool)) *testPeer { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + p := &testPeer{ln: ln, send: make(chan string, 8)} + go func() { + conn, err := ln.Accept() + if err != nil { + return + } + p.mu.Lock() + p.conn = conn + p.mu.Unlock() + l := link.New(conn, link.WithFramer(link.LengthPrefix(2))) + go func() { // peer-initiated frames + for s := range p.send { + _ = l.Send([]byte(s)) + } + }() + for { + raw, err := l.Receive() + if err != nil { + return + } + p.mu.Lock() + p.inbound = append(p.inbound, string(raw)) + silent := p.silent + p.mu.Unlock() + if silent { + continue + } + if out, ok := answer(string(raw)); ok { + if err := l.Send([]byte(out)); err != nil { + return + } + } + } + }() + t.Cleanup(func() { ln.Close() }) + return p +} + +func (p *testPeer) addr() string { return p.ln.Addr().String() } + +func (p *testPeer) goSilent() { + p.mu.Lock() + p.silent = true + p.mu.Unlock() +} + +// kill drops the accepted connection outright — the peer process dying, as distinct +// from goSilent (which keeps the socket open and simply stops answering). +func (p *testPeer) kill() { + p.mu.Lock() + conn := p.conn + p.mu.Unlock() + if conn != nil { + conn.Close() + } + p.ln.Close() +} + +func (p *testPeer) received() []string { + p.mu.Lock() + defer p.mu.Unlock() + return append([]string(nil), p.inbound...) +} + +func mtiOf(frame string) string { return strings.SplitN(frame, ":", 2)[0] } +func traceOf(frame string) string { + parts := strings.SplitN(frame, ":", 2) + if len(parts) != 2 { + return "" + } + return parts[1] +} + +// requestKeyer keys an OUTGOING frame by the response it expects (MTI function digit + 1). +func requestKeyer(frame []byte) (string, error) { + m, tr := mtiOf(string(frame)), traceOf(string(frame)) + if len(m) != 4 || tr == "" { + return "", errors.New("unkeyable") + } + return fmt.Sprintf("%s%c%s:%s", m[:2], m[2]+1, m[3:], tr), nil +} + +// responseKeyer keys an INBOUND frame by its own MTI. +func responseKeyer(frame []byte) (string, error) { + m, tr := mtiOf(string(frame)), traceOf(string(frame)) + if len(m) != 4 || tr == "" { + return "", errors.New("unkeyable") + } + return m + ":" + tr, nil +} + +// traceKeyer correlates on the trace alone, so a single keyer matches a request to its +// response across the MTI step. It is the shape that needs no ResponseKeyer — and the +// shape whose ambiguity ResponseKeyer exists to resolve. +func traceKeyer(frame []byte) (string, error) { + tr := traceOf(string(frame)) + if tr == "" { + return "", errors.New("unkeyable") + } + return tr, nil +} + +func dial(t *testing.T, cfg Config) *Connector { + t.Helper() + cfg.LinkOptions = append(cfg.LinkOptions, link.WithFramer(link.LengthPrefix(2))) + c, err := New(cfg) + if err != nil { + t.Fatal(err) + } + if err := c.Start(context.Background()); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { c.Stop(context.Background()) }) + return c +} + +func waitUp(t *testing.T, c *Connector, within time.Duration) { + t.Helper() + deadline := time.Now().Add(within) + for time.Now().Before(deadline) { + if c.Connected() { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatal("connector never came up") +} + +// --- gap 1: a peer-originated request can be ANSWERED ------------------------ + +func TestOnUnsolicited_CanAnswerAPeerOriginatedEcho(t *testing.T) { + peer := newTestPeer(t, func(frame string) (string, bool) { + if mtiOf(frame) == "0800" { // our keep-alive + return "0810:" + traceOf(frame), true + } + return "", false + }) + + answered := make(chan string, 1) + c := dial(t, Config{ + Name: "peer", Addr: peer.addr(), Keyer: responseKeyer, Timeout: time.Second, + OnUnsolicited: func(frame []byte, reply mux.Reply) { + // A peer echo-test: answer it, or the peer decides the link is dead. + if mtiOf(string(frame)) == "0800" { + out := "0810:" + traceOf(string(frame)) + if err := reply([]byte(out)); err == nil { + answered <- out + } + } + }, + }) + waitUp(t, c, 2*time.Second) + + peer.send <- "0800:777" // the PEER echo-tests US + + select { + case got := <-answered: + if got != "0810:777" { + t.Fatalf("answered %q, want 0810:777", got) + } + case <-time.After(2 * time.Second): + t.Fatal("a peer-originated 0800 was never answered") + } + + // And the answer actually reached the peer. + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + for _, f := range peer.received() { + if f == "0810:777" { + return + } + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("the peer never received our 0810; it saw %v", peer.received()) +} + +func TestSend_RequiresALiveLink(t *testing.T) { + c, err := New(Config{Name: "down", Addr: "127.0.0.1:1", Keyer: responseKeyer}) + if err != nil { + t.Fatal(err) + } + if err := c.Send([]byte("0810:1")); !errors.Is(err, ErrNotConnected) { + t.Fatalf("Send on a down link: %v, want ErrNotConnected", err) + } +} + +// --- gap 2: a peer request no longer satisfies our pending request ----------- + +func TestResponseKeyer_PeerRequestDoesNotSatisfyOurs(t *testing.T) { + // The peer never answers our 0800 — it only sends its OWN 0800 on the same trace. + peer := newTestPeer(t, func(frame string) (string, bool) { return "", false }) + + unsolicited := make(chan string, 4) + c := dial(t, Config{ + Name: "peer", Addr: peer.addr(), + Keyer: requestKeyer, // our 0800 registers under 0810: + ResponseKeyer: responseKeyer, // an inbound frame keys by its OWN mti + Timeout: 400 * time.Millisecond, + OnUnsolicited: func(frame []byte, _ mux.Reply) { unsolicited <- string(frame) }, + }) + waitUp(t, c, 2*time.Second) + + done := make(chan error, 1) + go func() { + _, err := c.Request(context.Background(), []byte("0800:555")) + done <- err + }() + time.Sleep(50 * time.Millisecond) + peer.send <- "0800:555" // same trace, but it is a REQUEST, not our answer + + select { + case got := <-unsolicited: + if got != "0800:555" { + t.Fatalf("unsolicited frame %q, want 0800:555", got) + } + case <-time.After(2 * time.Second): + t.Fatal("the peer's own 0800 was consumed as our reply instead of reaching the unsolicited handler") + } + + if err := <-done; !errors.Is(err, mux.ErrTimeout) { + t.Fatalf("our request resolved with %v; it must still time out — nothing answered it", err) + } +} + +// --- gap 3: a stalled keepalive no longer blinds the supervisor -------------- + +func TestKeepaliveTimeout_DetectsAHalfOpenPeerWithoutWaitingOutTheRequestTimeout(t *testing.T) { + // The peer answers until told to go silent, then holds the socket OPEN and stops + // answering: no FIN, no RST. Nothing ever closes the mux, so the only thing that can + // notice is the keepalive giving up. + peer := newTestPeer(t, func(frame string) (string, bool) { + return "0810:" + traceOf(frame), true + }) + + var stan atomic.Uint64 + c := dial(t, Config{ + Name: "peer", Addr: peer.addr(), Keyer: traceKeyer, + // A financial timeout: what an echo would inherit without KeepaliveTimeout. + Timeout: 30 * time.Second, + KeepaliveInterval: 50 * time.Millisecond, + KeepaliveTimeout: 200 * time.Millisecond, + Keepalive: func(ctx context.Context, m *mux.Mux) error { + _, err := m.Request(ctx, []byte(fmt.Sprintf("0800:%d", stan.Add(1)))) + return err + }, + MinBackoff: time.Minute, // do not let a reconnect mask the down state + MaxBackoff: time.Minute, + }) + waitUp(t, c, 2*time.Second) + + peer.goSilent() + + // interval + keepalive timeout + margin. Without KeepaliveTimeout this takes the full + // 30s request timeout, so a generous 5s here still fails against the old behaviour. + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if !c.Connected() { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal("a half-open peer left the connector reporting Connected(): the echo is still " + + "inheriting the financial request timeout, so live traffic keeps being written to a dead socket") +} + +func TestKeepalive_DoesNotBlockStop(t *testing.T) { + // A keepalive outstanding against a silent peer must not hold Stop hostage — off the + // supervise loop, ctx cancellation is observed immediately. + peer := newTestPeer(t, func(frame string) (string, bool) { + return "0810:" + traceOf(frame), true + }) + + var stan atomic.Uint64 + entered := make(chan struct{}, 1) + cfg := Config{ + Name: "peer", Addr: peer.addr(), Keyer: traceKeyer, + Timeout: 30 * time.Second, + KeepaliveInterval: 50 * time.Millisecond, + Keepalive: func(ctx context.Context, m *mux.Mux) error { + select { + case entered <- struct{}{}: + default: + } + _, err := m.Request(ctx, []byte(fmt.Sprintf("0800:%d", stan.Add(1)))) + return err + }, + LinkOptions: []link.Option{link.WithFramer(link.LengthPrefix(2))}, + } + c, err := New(cfg) + if err != nil { + t.Fatal(err) + } + if err := c.Start(context.Background()); err != nil { + t.Fatal(err) + } + waitUp(t, c, 2*time.Second) + peer.goSilent() + <-entered + + done := make(chan struct{}) + go func() { c.Stop(context.Background()); close(done) }() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("Stop blocked behind an outstanding keepalive") + } +} + +// --- gap 4: OnReady runs where Request actually works ------------------------ + +func TestOnReady_RunsAfterTheLinkIsUsable(t *testing.T) { + peer := newTestPeer(t, func(frame string) (string, bool) { + return "0810:" + traceOf(frame), true + }) + + ready := make(chan error, 1) + c := dial(t, Config{ + Name: "peer", Addr: peer.addr(), Keyer: traceKeyer, Timeout: time.Second, + OnReady: func(ctx context.Context, c *Connector) { + // The whole point: a ceremony issued through the CONNECTOR must work here. + // From OnConnect this would fail ErrNotConnected. + _, err := c.Request(ctx, []byte("0800:901")) + ready <- err + }, + }) + waitUp(t, c, 2*time.Second) + + select { + case err := <-ready: + if err != nil { + t.Fatalf("OnReady could not use the connector: %v", err) + } + case <-time.After(3 * time.Second): + t.Fatal("OnReady never ran") + } +} + +func TestOnConnect_CannotUseTheConnectorYet(t *testing.T) { + // Documents WHY OnReady exists: inside OnConnect the mux is not published. + peer := newTestPeer(t, func(frame string) (string, bool) { + return "0810:" + traceOf(frame), true + }) + + var self atomic.Pointer[Connector] + got := make(chan error, 1) + cfg := Config{ + Name: "peer", Addr: peer.addr(), Keyer: responseKeyer, Timeout: time.Second, + OnConnect: func(ctx context.Context, m *mux.Mux) error { + if c := self.Load(); c != nil { + select { + case got <- c.Send([]byte("0800:1")): + default: + } + } + return nil + }, + } + c, err := New(cfg) + if err != nil { + t.Fatal(err) + } + self.Store(c) + if err := c.Start(context.Background()); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { c.Stop(context.Background()) }) + + select { + case err := <-got: + if !errors.Is(err, ErrNotConnected) { + t.Fatalf("inside OnConnect the connector reported %v; expected ErrNotConnected", err) + } + case <-time.After(2 * time.Second): + t.Skip("OnConnect ran before the connector pointer was stored") + } +} diff --git a/mux/mux.go b/mux/mux.go index b151673..db2c233 100644 --- a/mux/mux.go +++ b/mux/mux.go @@ -42,9 +42,16 @@ var ( // and its response must match. type Keyer func(msg []byte) (string, error) +// Reply sends an answer to an unsolicited frame back over the same link. It is +// handed to an [WithUnsolicitedReplier] handler so a peer-initiated request — an +// echo test, a sign-off, a server-initiated advice — can be answered. +type Reply func(msg []byte) error + type config struct { timeout time.Duration unsolicited func([]byte) + replier func([]byte, Reply) + respKey Keyer } // Option configures a Mux. @@ -56,10 +63,42 @@ func WithTimeout(d time.Duration) Option { return func(c *config) { c.timeout = // WithUnsolicitedHandler sets a handler for frames that match no in-flight // request (server-initiated requests, late responses). It runs on the reader // goroutine, so it must not block. +// +// Use [WithUnsolicitedReplier] instead when the handler needs to ANSWER the frame: +// a peer that echo-tests us and never receives a reply concludes the link is dead +// and tears it down. func WithUnsolicitedHandler(h func([]byte)) Option { return func(c *config) { c.unsolicited = h } } +// WithUnsolicitedReplier sets an unsolicited handler that can answer the frame, via +// the [Reply] it is given. It takes precedence over [WithUnsolicitedHandler]. +// +// Like the plain handler it runs on the reader goroutine and must not block — but +// a reply is a single non-blocking write to the link, so answering inline is the +// intended use. Anything slower belongs on a queue. +func WithUnsolicitedReplier(h func(frame []byte, reply Reply)) Option { + return func(c *config) { c.replier = h } +} + +// WithResponseKeyer sets a SEPARATE keyer for inbound frames, leaving the Keyer +// passed to [New] to key outgoing requests only. Without it one function keys both +// directions, and then no keyer can both match a response to its request and tell a +// PEER's request apart from our own — the two are structurally identical. +// +// That ambiguity is not theoretical. Network-management traffic correlates on the +// trace number alone (an 0800 echo carries no terminal id), so a peer-originated +// 0800 whose trace happens to match one we have in flight is delivered to us as +// though it were the 0810 answer, and the real answer then arrives unmatched. +// +// With both keyers the caller resolves it by keying on the message type as well as +// the trace: the request keyer maps an outgoing frame to the key of the response it +// EXPECTS, and the response keyer maps an inbound frame by its OWN type. Our 0800 +// registers under "0810:", the real 0810 matches it, and a peer's 0800 keys +// to "0800:" — no match, so it reaches the unsolicited handler where it +// belongs. +func WithResponseKeyer(k Keyer) Option { return func(c *config) { c.respKey = k } } + // Mux correlates requests and responses over a link. type Mux struct { link *link.Link @@ -137,7 +176,7 @@ func (m *Mux) readLoop() { m.fail(err) return } - k, kerr := m.key(msg) + k, kerr := m.responseKey(msg) if kerr != nil { m.deliverUnsolicited(msg) continue @@ -156,12 +195,40 @@ func (m *Mux) readLoop() { } } +// responseKey keys an INBOUND frame, using the response keyer when one is set and +// otherwise the request keyer (the single-keyer behaviour). +func (m *Mux) responseKey(msg []byte) (string, error) { + if m.cfg.respKey != nil { + return m.cfg.respKey(msg) + } + return m.key(msg) +} + func (m *Mux) deliverUnsolicited(msg []byte) { + if m.cfg.replier != nil { + m.cfg.replier(msg, m.Send) + return + } if m.cfg.unsolicited != nil { m.cfg.unsolicited(msg) } } +// Send writes msg to the link WITHOUT registering a correlation or waiting for a +// response. It is how a peer-initiated frame is answered — an echo test, a sign-off, +// an advice — where [Mux.Request] would be wrong: the frame we are sending IS the +// response, so there is nothing to wait for. +// +// Safe for concurrent use (the link serialises writes). +func (m *Mux) Send(msg []byte) error { + select { + case <-m.done: + return m.closedErr() + default: + } + return m.link.Send(msg) +} + // stop closes the mux once, recording cause for Err. The first caller wins, so a // clean Close reports ErrClosed while a read-loop failure reports its real cause // even if Close races in just behind it.