From cfa20f05ad6d71db3d458a50ec6b2e68570af8f2 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 04:52:10 +0200 Subject: [PATCH] Reconnect without limit, as the tool runner does, so a process outlives a bus restart (issue 1192, review round 1 finding 5) --- go/process/conn.go | 37 ++++++++++++++++++++++-------------- go/process/reconnect_test.go | 27 ++++++++++++++++++++++++++ 2 files changed, 50 insertions(+), 14 deletions(-) create mode 100644 go/process/reconnect_test.go diff --git a/go/process/conn.go b/go/process/conn.go index 0c31c4c..50addf5 100644 --- a/go/process/conn.go +++ b/go/process/conn.go @@ -38,15 +38,30 @@ var refusedPublish = regexp.MustCompile(`(?i)Publish to "(\S+)"`) // Connect dials the bus with account: pinned to its fingerprint when it carries one (refused with any // other certificate), authenticated as its user and password, on the inbox prefix _INBOX. and no -// other. Ported from mesh-controller internal/broker/jetstream.go:106-132 (DialPinned, -// PinnedToFingerprint) since this package may not import the controller. +// other, reconnecting without limit as the tool runner does (novox/hq issue 1192: a process outlives a +// bus restart, never giving up the way the client library's own default of 60 attempts would). Ported +// from mesh-controller internal/broker/jetstream.go:106-132 (DialPinned, PinnedToFingerprint) since this +// package may not import the controller. // // An ErrorHandler among opts is chained behind this package's own, which watches every async error for a // permission violation refusing a publish this connection is waiting an answer to, and cancels that wait // at once. func Connect(account Account, opts ...nats.Option) (*Conn, error) { c := &Conn{account: account, pending: map[string]map[uint64]context.CancelCauseFunc{}} + nc, err := nats.Connect(account.URL, c.options(opts)...) + if err != nil { + return nil, fmt.Errorf("connecting the process account %s: %w", account.User, err) + } + c.nc = nc + return c, nil +} +// options is the full option list Connect dials with, apart from Connect so a unit test can check what +// it carries without a live bus: this package's own (the inbox prefix, the credentials, the pin, +// reconnecting without limit, and an ErrorHandler chaining any the caller passed), then opts, then this +// package's own ErrorHandler again last, so it is the one AsyncErrorCB nats.Connect keeps however opts +// ordered theirs. +func (c *Conn) options(opts []nats.Option) []nats.Option { // opts are functions over a *nats.Options; applied to a scratch value, this reads off a caller's own // ErrorHandler (if any) to chain, without needing nats.Connect's own URL/Servers parsing here. var scratch nats.Options @@ -58,26 +73,20 @@ func Connect(account Account, opts ...nats.Option) (*Conn, error) { chained := scratch.AsyncErrorCB all := []nats.Option{ - nats.CustomInboxPrefix(account.InboxPrefix()), - nats.UserInfo(account.User, account.Password), + nats.CustomInboxPrefix(c.account.InboxPrefix()), + nats.UserInfo(c.account.User, c.account.Password), + nats.MaxReconnects(-1), } - if account.Fingerprint != "" { - all = append(all, nats.Secure(pinnedToFingerprint(account.Fingerprint))) + if c.account.Fingerprint != "" { + all = append(all, nats.Secure(pinnedToFingerprint(c.account.Fingerprint))) } all = append(all, opts...) - // Last, so it is the one AsyncErrorCB nats.Connect keeps however the opts above ordered theirs. - all = append(all, nats.ErrorHandler(func(nc *nats.Conn, sub *nats.Subscription, err error) { + return append(all, nats.ErrorHandler(func(nc *nats.Conn, sub *nats.Subscription, err error) { c.publishRefused(err) if chained != nil { chained(nc, sub, err) } })) - nc, err := nats.Connect(account.URL, all...) - if err != nil { - return nil, fmt.Errorf("connecting the process account %s: %w", account.User, err) - } - c.nc = nc - return c, nil } // Close ends the connection. diff --git a/go/process/reconnect_test.go b/go/process/reconnect_test.go new file mode 100644 index 0000000..03a7ffd --- /dev/null +++ b/go/process/reconnect_test.go @@ -0,0 +1,27 @@ +package process + +import ( + "testing" + + "github.com/nats-io/nats.go" +) + +// novox/hq issue 1192 (taken into this round, release 2's reviewer): Connect reconnects without limit, as +// the tool runner does, so a process outlives a bus restart instead of giving up after the client +// library's default of 60 attempts. Checked without a live bus: the option list Connect dials with is +// applied to a scratch nats.Options and MaxReconnect read back. +func TestConnectOptionsReconnectWithoutLimit(t *testing.T) { + c := &Conn{account: Account{Node: "novox", Module: "mesh-mail", User: "novox.mesh-mail.send", Password: "pw"}} + var o nats.Options + for _, opt := range c.options(nil) { + if opt == nil { + continue + } + if err := opt(&o); err != nil { + t.Fatal(err) + } + } + if o.MaxReconnect != -1 { + t.Errorf("MaxReconnect: %d, want -1 (no limit)", o.MaxReconnect) + } +}