From 35b779f3634294f3fad642a63fb1bc9487e33498 Mon Sep 17 00:00:00 2001 From: Canaan Etai Date: Fri, 11 Sep 2026 16:53:27 +0100 Subject: [PATCH 1/2] feat(connector,mux): close four network-management gaps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Found while tracing ISW/ZONE link instability in a downstream acquirer. Each gap forced a workaround that could not fully fix the problem from outside the library. Reply to an unsolicited frame. Unsolicited handed over the frame and nothing else, and neither Mux nor Connector exposed a raw Send — so a peer-initiated message could not be answered at all. Postilion-family hosts do not only answer echoes, they send them, and a host that echo-tests us and never receives an 0810 concludes the link is dead and tears it down. Adds mux.Reply, WithUnsolicitedReplier, (*Mux).Send, (*Connector).Send and Config.OnUnsolicited. Direction-blind Keyer. One keyer served both directions, and with a single function no keyer can match a response to its request and also separate a peer's request from our own — the two are structurally identical. Network management correlates on the trace alone, so a peer 0800 on a colliding trace was delivered as our 0810 while the real answer arrived unmatched. Adds WithResponseKeyer / Config.ResponseKeyer: 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. Keepalive inheriting the request timeout. A peer that dies cleanly unblocks the pending keepalive through the mux, so that case was always handled. 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 traffic onto a dead socket. Adds Config.KeepaliveTimeout, bounding one keepalive independently. serve also runs the keepalive off the supervise loop, which does not shrink that window but stops an outstanding keepalive holding Stop hostage. No hook after the mux is published. OnConnect runs before the mux is stored, which is right for sign-on but leaves nowhere for a ceremony that must go through the connector's own request path — it fails ErrNotConnected. Adds Config.OnReady. All additive: existing configurations behave exactly as before. Six tests, including one that fails without KeepaliveTimeout and one documenting why OnReady exists. --- connector/connector.go | 99 ++++++++- connector/netmgmt_test.go | 412 ++++++++++++++++++++++++++++++++++++++ mux/mux.go | 69 ++++++- 3 files changed, 575 insertions(+), 5 deletions(-) create mode 100644 connector/netmgmt_test.go 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. From 2ea34eac336151634cf3d049082712abbe322cf8 Mon Sep 17 00:00:00 2001 From: Canaan Etai Date: Fri, 11 Sep 2026 17:08:45 +0100 Subject: [PATCH 2/2] =?UTF-8?q?docs(changelog):=20record=201.1.0=20?= =?UTF-8?q?=E2=80=94=20network-management=20additions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 51 ++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 50 insertions(+), 1 deletion(-) 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