From 131e9ff0935d0b55640771770bae1b141f68a5b5 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Sat, 12 Sep 2026 05:40:43 +0200 Subject: [PATCH] feat: give the JetStream stream a size ceiling from NATS_MAX_BYTES, accepting a byte count or a size like 2GiB, because a managed account can require every stream to declare one and Synadia's Max Bytes Required rejects creation without it, turning that refusal into an error naming the variable to set, and fix the retry path which passed a zero max age and would have recreated the stream with no age limit after a failure --- .../docs/development/configuration.mdx | 1 + internal/api/handler/fleet_nodes.go | 1 + internal/infrastructure/eventbus/factory.go | 5 ++ internal/infrastructure/eventbus/nats.go | 47 +++++++++++++-- .../infrastructure/eventbus/nats_maxbytes.go | 57 +++++++++++++++++++ .../eventbus/nats_maxbytes_test.go | 53 +++++++++++++++++ 6 files changed, 158 insertions(+), 6 deletions(-) create mode 100644 internal/infrastructure/eventbus/nats_maxbytes.go create mode 100644 internal/infrastructure/eventbus/nats_maxbytes_test.go diff --git a/docs/content/docs/development/configuration.mdx b/docs/content/docs/development/configuration.mdx index e7ef394e..7fbe3ae2 100644 --- a/docs/content/docs/development/configuration.mdx +++ b/docs/content/docs/development/configuration.mdx @@ -293,6 +293,7 @@ On `filesystem`, a remote worker writes blobs to its own disk rather than a volu | `NATS_URL` | JetStream address. Credentials in the URL are honored by every service, including the Rust tracking publisher: `nats://user:pass@host:4222` for a user, `nats://token@host:4222` for a token, `tls://` for TLS | `nats://nats:4222` | yes | | `NATS_CREDS` | Path to a NATS credentials file (user JWT plus nkey seed), for a bus that authenticates with JWT rather than a token. Synadia Cloud and any nsc-managed account issue one | unset | yes | | `NATS_CREDS_B64` | The same file, base64 encoded, as a single line. This is the form the fleet uses: a node receives environment variables rather than files, and the env file docker reads cannot express a multi-line value. Takes precedence over `NATS_CREDS` | unset | yes | +| `NATS_MAX_BYTES` | Ceiling on the stream's size on disk. Accepts a plain byte count or a size (`2GiB`, `512MB`, `1G`). Unset leaves the stream bounded only by `NATS_STREAM_MAX_AGE` and the account's own quota, which a managed bus may refuse: Synadia's "Max Bytes Required" rejects any stream created without one | unset (unlimited) | yes | | `NATS_STREAM_NAME`, `NATS_SUBJECT_PREFIX` | Stream and subject naming | `warmbly` | yes | | `KAFKA_BOOTSTRAP_SERVERS` | Broker list when `EVENTBUS_PROVIDER=kafka` | unset | yes | | `KAFKA_SASL_USERNAME`, `KAFKA_SASL_PASSWORD` | Broker credentials | unset | yes | diff --git a/internal/api/handler/fleet_nodes.go b/internal/api/handler/fleet_nodes.go index 6cbb86ff..bb79744b 100644 --- a/internal/api/handler/fleet_nodes.go +++ b/internal/api/handler/fleet_nodes.go @@ -220,6 +220,7 @@ var nodeEnvKeys = []string{ // Base64, not a path: a node has no file to point at, and the env file // docker reads cannot hold a multi-line value. "NATS_CREDS_B64", + "NATS_MAX_BYTES", "KAFKA_BOOTSTRAP_SERVERS", "KAFKA_SASL_USERNAME", "KAFKA_SASL_PASSWORD", diff --git a/internal/infrastructure/eventbus/factory.go b/internal/infrastructure/eventbus/factory.go index ec4bc16d..bb4f6507 100644 --- a/internal/infrastructure/eventbus/factory.go +++ b/internal/infrastructure/eventbus/factory.go @@ -46,10 +46,15 @@ func FromEnv(bootstrap string, sasl *kafka.SASLConfig) (EventBus, error) { SASL: sasl, }) case "nats": + maxBytes, err := parseByteSize(os.Getenv("NATS_MAX_BYTES")) + if err != nil { + return nil, err + } return NewNATS(NATSConfig{ URL: natsURLFromEnv(), StreamName: os.Getenv("NATS_STREAM_NAME"), SubjectPrefix: os.Getenv("NATS_SUBJECT_PREFIX"), + MaxBytes: maxBytes, }) default: return nil, fmt.Errorf("eventbus: unknown EVENTBUS_PROVIDER %q (want: kafka, nats)", provider) diff --git a/internal/infrastructure/eventbus/nats.go b/internal/infrastructure/eventbus/nats.go index bcfa740f..e839e0ac 100644 --- a/internal/infrastructure/eventbus/nats.go +++ b/internal/infrastructure/eventbus/nats.go @@ -34,6 +34,11 @@ type NATSBus struct { stream string prefix string + // Held rather than passed: ensureStream retries after a failure, and the + // retry callers have no config to hand it. + maxAge time.Duration + maxBytes int64 + mu sync.Mutex subscribers []jetstream.ConsumeContext streamEnsure sync.Once @@ -59,6 +64,11 @@ type NATSConfig struct { // "use the stream's existing setting or 7 days for new streams". MaxAge time.Duration + // MaxBytes caps the stream on disk. Zero leaves it unbounded, which a + // managed account may refuse: Synadia's "Max Bytes Required" rejects any + // stream created without one. + MaxBytes int64 + // Options passed to nats.Connect (auth, TLS, etc). Options []nats.Option } @@ -110,11 +120,14 @@ func NewNATS(cfg NATSConfig) (*NATSBus, error) { js: js, stream: cfg.StreamName, prefix: cfg.SubjectPrefix, + + maxAge: cfg.MaxAge, + maxBytes: cfg.MaxBytes, } // Eagerly ensure the stream so misconfiguration surfaces at boot rather // than on the first publish. Failures are non-fatal here — the lazy // retry inside Publish/Subscribe will surface them to callers. - if err := b.ensureStream(context.Background(), cfg.MaxAge); err != nil { + if err := b.ensureStream(context.Background()); err != nil { log.Warn().Err(err).Msg("eventbus nats: deferred stream setup") } return b, nil @@ -152,7 +165,7 @@ func (b *NATSBus) durable(group string, topics []string) string { return sb.String() } -func (b *NATSBus) ensureStream(ctx context.Context, maxAge time.Duration) error { +func (b *NATSBus) ensureStream(ctx context.Context) error { b.streamEnsure.Do(func() { cctx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() @@ -162,11 +175,12 @@ func (b *NATSBus) ensureStream(ctx context.Context, maxAge time.Duration) error Subjects: []string{filter}, Retention: jetstream.LimitsPolicy, Storage: jetstream.FileStorage, - MaxAge: maxAge, + MaxAge: b.maxAge, + MaxBytes: maxBytesOrUnlimited(b.maxBytes), Discard: jetstream.DiscardOld, }) if err != nil { - b.streamErr = fmt.Errorf("eventbus nats: ensure stream %q: %w", b.stream, err) + b.streamErr = fmt.Errorf("eventbus nats: ensure stream %q: %w%s", b.stream, err, maxBytesHint(err, b.maxBytes)) b.streamEnsure = sync.Once{} // allow retry on next call } }) @@ -184,7 +198,7 @@ func (b *NATSBus) Publish(ctx context.Context, topic, key string, payload []byte } b.mu.Unlock() - if err := b.ensureStream(ctx, 0); err != nil { + if err := b.ensureStream(ctx); err != nil { return err } @@ -218,7 +232,7 @@ func (b *NATSBus) Subscribe(ctx context.Context, topics []string, group string, if handler == nil { return errors.New("eventbus nats: handler required") } - if err := b.ensureStream(ctx, 0); err != nil { + if err := b.ensureStream(ctx); err != nil { return err } @@ -334,3 +348,24 @@ func natsURLFromEnv() string { // Compile-time interface check. var _ EventBus = (*NATSBus)(nil) + +// maxBytesOrUnlimited maps an unset ceiling onto JetStream's own spelling for +// it. Zero in a StreamConfig means zero bytes, not unlimited. +func maxBytesOrUnlimited(v int64) int64 { + if v <= 0 { + return -1 + } + return v +} + +// maxBytesHint turns the server's refusal into the thing to do about it. The +// bare error names a policy the operator has probably never heard of. +func maxBytesHint(err error, maxBytes int64) string { + if maxBytes > 0 || err == nil { + return "" + } + if !strings.Contains(strings.ToLower(err.Error()), "max bytes") { + return "" + } + return " (this account requires every stream to declare a size; set NATS_MAX_BYTES, e.g. NATS_MAX_BYTES=1GiB)" +} diff --git a/internal/infrastructure/eventbus/nats_maxbytes.go b/internal/infrastructure/eventbus/nats_maxbytes.go new file mode 100644 index 00000000..a111340d --- /dev/null +++ b/internal/infrastructure/eventbus/nats_maxbytes.go @@ -0,0 +1,57 @@ +package eventbus + +import ( + "fmt" + "strconv" + "strings" +) + +// parseByteSize reads NATS_MAX_BYTES. Plain bytes, or a size with a suffix: +// 2GiB, 512MB, 1G. Binary and decimal are both accepted because operators +// reach for either, and a stream ceiling does not need the distinction to be +// load bearing. +// +// Empty means unset, which leaves the stream bounded only by MaxAge and the +// account's own quota. +func parseByteSize(raw string) (int64, error) { + s := strings.TrimSpace(raw) + if s == "" { + return 0, nil + } + upper := strings.ToUpper(s) + + // Longest suffix first: GIB has to match before GB, and GB before G. + units := []struct { + suffix string + mult int64 + }{ + {"TIB", 1 << 40}, {"GIB", 1 << 30}, {"MIB", 1 << 20}, {"KIB", 1 << 10}, + {"TB", 1e12}, {"GB", 1e9}, {"MB", 1e6}, {"KB", 1e3}, + {"T", 1 << 40}, {"G", 1 << 30}, {"M", 1 << 20}, {"K", 1 << 10}, + {"B", 1}, + } + for _, u := range units { + if !strings.HasSuffix(upper, u.suffix) { + continue + } + num := strings.TrimSpace(upper[:len(upper)-len(u.suffix)]) + // Fractional sizes are the natural way to write 2.5GiB. + f, err := strconv.ParseFloat(num, 64) + if err != nil { + return 0, fmt.Errorf("eventbus nats: NATS_MAX_BYTES %q is not a size", raw) + } + if f < 0 { + return 0, fmt.Errorf("eventbus nats: NATS_MAX_BYTES %q is negative", raw) + } + return int64(f * float64(u.mult)), nil + } + + n, err := strconv.ParseInt(upper, 10, 64) + if err != nil { + return 0, fmt.Errorf("eventbus nats: NATS_MAX_BYTES %q is not a size", raw) + } + if n < 0 { + return 0, fmt.Errorf("eventbus nats: NATS_MAX_BYTES %q is negative", raw) + } + return n, nil +} diff --git a/internal/infrastructure/eventbus/nats_maxbytes_test.go b/internal/infrastructure/eventbus/nats_maxbytes_test.go new file mode 100644 index 00000000..4e96ff15 --- /dev/null +++ b/internal/infrastructure/eventbus/nats_maxbytes_test.go @@ -0,0 +1,53 @@ +package eventbus + +import "testing" + +// The ceiling is written by hand in a dashboard or an env file, so the shapes +// an operator actually types have to parse: a bare count, binary and decimal +// suffixes, and the fractional form a console shows (2.5 GiB). +func TestParseByteSize(t *testing.T) { + cases := map[string]int64{ + "": 0, + "1024": 1024, + "1KiB": 1024, + "1 KiB": 1024, + "2GiB": 2 << 30, + "2.5GiB": int64(2.5 * float64(1<<30)), + "512MB": 512_000_000, + "1G": 1 << 30, + "1B": 1, + " 4MiB ": 4 << 20, + "1gib": 1 << 30, + } + for in, want := range cases { + got, err := parseByteSize(in) + if err != nil { + t.Errorf("%q: unexpected error %v", in, err) + continue + } + if got != want { + t.Errorf("%q = %d, want %d", in, got, want) + } + } +} + +// A typo must stop the process at boot, not silently leave the stream +// unbounded on an account that will then refuse to create it. +func TestParseByteSizeRejectsNonsense(t *testing.T) { + for _, in := range []string{"big", "1XB", "-5", "-1GiB", "1.2.3GiB"} { + if _, err := parseByteSize(in); err == nil { + t.Errorf("%q was accepted", in) + } + } +} + +// JetStream spells "no ceiling" as -1; a literal 0 in a StreamConfig means +// zero bytes, which would reject every message. +func TestMaxBytesOrUnlimited(t *testing.T) { + if got := maxBytesOrUnlimited(0); got != -1 { + t.Errorf("unset = %d, want -1", got) + } + if got := maxBytesOrUnlimited(1 << 30); got != 1<<30 { + t.Errorf("set = %d, want %d", got, 1<<30) + } +}