|
|
@@ -70,21 +70,33 @@ func (c *Client) NC() *nats.Conn { return c.nc }
|
|
|
|
|
|
// EnsureStreams creates the three JetStream streams if they don't
|
|
|
// exist. M0: single-node, no replication.
|
|
|
+//
|
|
|
+// Retention: tuned for the M11 NATS resource-limit finding. The
|
|
|
+// 24h MaxAge on ALERTS allowed 6+ GiB of test data to accumulate
|
|
|
+// and exceed the server-level max_storage cap. With a 1h MaxAge
|
|
|
+// plus a 1 GiB MaxBytes safety cap, the stream self-trims long
|
|
|
+// before the server cap is hit. routerd is the only consumer and
|
|
|
+// processes in real time, so 1h is a generous safety window.
|
|
|
+//
|
|
|
+// See M11_NATS_INVESTIGATION.md for the full analysis.
|
|
|
func (c *Client) EnsureStreams(ctx context.Context) error {
|
|
|
streams := []struct {
|
|
|
name string
|
|
|
subjects []string
|
|
|
age time.Duration
|
|
|
+ maxBytes int64
|
|
|
}{
|
|
|
- {"ALERTS", []string{"alerts.>"}, 24 * time.Hour},
|
|
|
- {"DELIVERIES", []string{"deliveries.>"}, 1 * time.Hour},
|
|
|
- {"DLQ", []string{"dlq.>"}, 7 * 24 * time.Hour},
|
|
|
+ {"ALERTS", []string{"alerts.>"}, 1 * time.Hour, 1 << 30}, // 1h, 1 GiB
|
|
|
+ {"DELIVERIES", []string{"deliveries.>"}, 1 * time.Hour, 100 << 20}, // 1h, 100 MiB
|
|
|
+ {"DLQ", []string{"dlq.>"}, 1 * time.Hour, 10 << 20}, // 1h, 10 MiB
|
|
|
}
|
|
|
for _, s := range streams {
|
|
|
_, err := c.js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
|
|
|
Name: s.name,
|
|
|
Subjects: s.subjects,
|
|
|
MaxAge: s.age,
|
|
|
+ MaxBytes: s.maxBytes,
|
|
|
+ Discard: jetstream.DiscardOld,
|
|
|
Storage: jetstream.FileStorage,
|
|
|
})
|
|
|
if err != nil {
|