fix: add per-stream MaxBytes/MaxAge config for LLMLog stream (#1246)

Co-authored-by: Jun Luo <luojun96@live.cn>

Author: Rader

builder/mq/nasts.go

@@ -103,7 +103,13 @@ func (n *Nats) getOrCreateStreamConsumer(params SubscribeParams) (jetstream.Cons
 			Subjects:     params.Topics,
 			MaxConsumers: -1,
 			MaxMsgs:      -1,
-			MaxBytes:     -1,
+			// NATS JetStream's unlimited sentinel for byte limit is -1.
+			// Values <= 0 from SubscribeParams are mapped to -1.
+			// See https://nats-io.github.io/nats.net/api/NATS.Client.JetStream.Models.StreamConfig.html
+			MaxBytes: maxBytesOrDefault(params.MaxBytes),
+			// NATS JetStream treats MaxAge 0 as unlimited (no age-based expiry).
+			// See https://nats-io.github.io/nats.net/api/NATS.Client.JetStream.Models.StreamConfig.MaxAge.html
+			MaxAge: params.MaxAge,
 		})
 	if err != nil {
 		return nil, fmt.Errorf("[nats] failed to create or update stream %s error: %w", params.Group.StreamName, err)
@@ -211,3 +217,10 @@ func (n *Nats) DeleteMessagesByFilter(streamName string, filter func(data []byte
 		slog.Int("deleted_count", deletedCount))
 	return nil
 }
+
+func maxBytesOrDefault(v int64) int64 {
+	if v <= 0 {
+		return -1
+	}
+	return v
+}

builder/mq/types.go

@@ -1,5 +1,7 @@
 package mq
 
+import "time"
+
 const (
 	NotifySubChangeSubject  string = "accounting.notify.subscription"
 	NotifyStopDeploySubject string = "accounting.notify.stopdeploy"
@@ -129,4 +131,10 @@ type SubscribeParams struct {
 	AutoACK                bool // auto acknowledge message after callback success
 	IsRedeliverForCBFailed bool // whether or not redeliver message for callback return error
 	Callback               MessageCallback
+	// MaxBytes sets the stream's maximum size in bytes.
+	// Non-positive values are handled by the message queue implementation.
+	MaxBytes int64
+	// MaxAge sets the stream's maximum message age.
+	// Zero value is handled by the message queue implementation.
+	MaxAge time.Duration
 }

common/config/config.go

@@ -514,6 +514,8 @@ type Config struct {
 		WorkerNum            int    `env:"OPENCSG_LLMLOG_WORKER_NUM" default:"10"`
 		BatchSize            int    `env:"OPENCSG_LLMLOG_BATCH_SIZE" default:"1000"`
 		FlushIntervalSeconds int    `env:"OPENCSG_LLMLOG_FLUSH_INTERVAL_SECONDS" default:"300"`
+		StreamMaxBytes       int64  `env:"OPENCSG_LLMLOG_STREAM_MAX_BYTES" default:"8589934592"`    // 8 GiB
+		StreamMaxAgeSeconds  int    `env:"OPENCSG_LLMLOG_STREAM_MAX_AGE_SECONDS" default:"1209600"` // 14 days
 		Sync                 struct {
 			Enable            bool   `env:"STARHUB_SERVER_LLMLOG_SYNC_ENABLE" default:"false"`
 			Username          string `env:"STARHUB_SERVER_LLMLOG_SYNC_USERNAME" default:""`