Merge pull request #446 from warmbly/feat/nats-max-bytes

feat: a stream size ceiling, and a latent bug in the retry path

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 |

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",

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)

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

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

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