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) + } +}