From fb8998c3290ca706218b85b67662d3e28abc0ffe Mon Sep 17 00:00:00 2001 From: archandatta <35818003+archandatta@users.noreply.github.com> Date: Fri, 18 Sep 2026 11:18:09 +0000 Subject: [PATCH 1/7] Add S2StorageController to own the S2 sink lifecycle The writer is opened at boot today, which leaves no way to decide per session whether events are persisted at all. Wrap it in a controller that resolves the stream lazily and opens at most one writer, so a later change can start it from the telemetry handler instead. Co-Authored-By: Claude Opus 5 --- server/lib/events/s2storage.go | 82 ++++++++++++++++++++++++++++++++++ 1 file changed, 82 insertions(+) diff --git a/server/lib/events/s2storage.go b/server/lib/events/s2storage.go index d10b51f54..c0e6c1b67 100644 --- a/server/lib/events/s2storage.go +++ b/server/lib/events/s2storage.go @@ -226,3 +226,85 @@ func (w *S2StorageWriter) Stop(ctx context.Context) error { } return w.storage.Close(ctx) } + +// S2StorageController owns the lifecycle of the S2 sink so the writer can be +// opened on demand rather than at boot. The append session binds one stream for +// its lifetime and the writer under it is single-use, so the controller opens at +// most one writer: once a writer has opened, Start is a no-op for the rest of +// the process. Safe for concurrent use. +type S2StorageController struct { + es *EventStream + basin string + token string + streamFn func() string + cfg S2Config + log *slog.Logger + + mu sync.Mutex + writer *S2StorageWriter + cancel context.CancelFunc + everStarted bool +} + +// NewS2StorageController resolves the stream name through streamFn at Start +// rather than here: an instance holding for a fork identity carries the stream +// of the instance it was forked from until the handoff lands. +func NewS2StorageController(es *EventStream, basin, token string, streamFn func() string, cfg S2Config, log *slog.Logger) *S2StorageController { + return &S2StorageController{es: es, basin: basin, token: token, streamFn: streamFn, cfg: cfg, log: log} +} + +// Start opens the sink, or is a no-op when a writer has already opened or the +// basin, token, or stream is unset. A failed open leaves the controller unstarted +// so a later identity can still open one. parent governs the read loop; the +// controller derives a cancelable child so Stop can halt the loop even when +// parent is still live. +func (c *S2StorageController) Start(parent context.Context) error { + c.mu.Lock() + defer c.mu.Unlock() + if c.everStarted { + return nil + } + stream := c.streamFn() + if c.basin == "" || c.token == "" || stream == "" { + return nil + } + c.log.Info("S2 storage enabled", "basin", c.basin, "stream", stream) + runCtx, cancel := context.WithCancel(parent) + w := NewS2StorageWriter(c.es, c.basin, c.token, stream, c.cfg, c.log) + if err := w.Start(runCtx); err != nil { + cancel() + return err + } + c.writer, c.cancel, c.everStarted = w, cancel, true + return nil +} + +// Stop drains and shuts down a running writer, or is a no-op if none is running. +// ctx bounds shutdown time. +func (c *S2StorageController) Stop(ctx context.Context) error { + c.mu.Lock() + defer c.mu.Unlock() + if c.writer == nil { + return nil + } + c.cancel() + err := c.writer.Stop(ctx) + c.writer, c.cancel = nil, nil + return err +} + +// Running reports whether the sink is currently forwarding events. +func (c *S2StorageController) Running() bool { + c.mu.Lock() + defer c.mu.Unlock() + return c.writer != nil +} + +// EverStarted reports whether a writer ever opened, including one already +// stopped. A caller that must guarantee nothing was persisted needs this rather +// than Running, which goes false again at shutdown. +func (c *S2StorageController) EverStarted() bool { + c.mu.Lock() + defer c.mu.Unlock() + return c.everStarted +} From 21a5399d4fd6d29675826c59af5bca3596de7c05 Mon Sep 17 00:00:00 2001 From: archandatta <35818003+archandatta@users.noreply.github.com> Date: Fri, 18 Sep 2026 11:19:40 +0000 Subject: [PATCH 2/7] Test that the S2 controller opens nothing until Start Co-Authored-By: Claude Opus 5 --- server/lib/events/s2storage_test.go | 66 +++++++++++++++++++++++++++++ 1 file changed, 66 insertions(+) create mode 100644 server/lib/events/s2storage_test.go diff --git a/server/lib/events/s2storage_test.go b/server/lib/events/s2storage_test.go new file mode 100644 index 000000000..0fd8c08e8 --- /dev/null +++ b/server/lib/events/s2storage_test.go @@ -0,0 +1,66 @@ +package events + +import ( + "context" + "log/slog" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// newTestController builds a controller whose streamFn returns stream and +// records how many times it was called, so a test can assert the name is +// resolved once, at Start. +func newTestController(t *testing.T, stream string) (*S2StorageController, *atomic.Int32) { + t.Helper() + var resolved atomic.Int32 + c := NewS2StorageController(newTestStream(t, 64), "test-basin", "test-token", func() string { + resolved.Add(1) + return stream + }, S2Config{}, slog.Default()) + return c, &resolved +} + +func stopController(t *testing.T, c *S2StorageController) error { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + return c.Stop(ctx) +} + +// TestS2StorageController_OpensNothingUntilStart is the guarantee the whole +// export-only mode rests on: a controller that is never started resolves no +// stream, so no append session is opened and nothing is persisted. +func TestS2StorageController_OpensNothingUntilStart(t *testing.T) { + c, resolved := newTestController(t, "test-stream") + + assert.Zero(t, resolved.Load(), "stream name must not be resolved before Start") + assert.False(t, c.Running()) + assert.False(t, c.EverStarted()) +} + +func TestS2StorageController_StartIsIdempotent(t *testing.T) { + c, resolved := newTestController(t, "test-stream") + + require.NoError(t, c.Start(context.Background())) + first := c.writer + require.NotNil(t, first) + + require.NoError(t, c.Start(context.Background())) + assert.Same(t, first, c.writer, "second Start must not replace the writer") + assert.Equal(t, int32(1), resolved.Load(), "second Start must not re-resolve the stream") + + require.NoError(t, stopController(t, c)) +} + +func TestS2StorageController_EmptyStreamDoesNotStart(t *testing.T) { + c, resolved := newTestController(t, "") + + require.NoError(t, c.Start(context.Background())) + assert.Equal(t, int32(1), resolved.Load()) + assert.False(t, c.Running()) + assert.False(t, c.EverStarted()) +} From fa1ae28391c3c9dd6bc9b02410239da65e89723a Mon Sep 17 00:00:00 2001 From: archandatta <35818003+archandatta@users.noreply.github.com> Date: Fri, 18 Sep 2026 11:19:43 +0000 Subject: [PATCH 3/7] Test the S2 controller stop path Co-Authored-By: Claude Opus 5 --- server/lib/events/s2storage_test.go | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/server/lib/events/s2storage_test.go b/server/lib/events/s2storage_test.go index 0fd8c08e8..ab4a35d96 100644 --- a/server/lib/events/s2storage_test.go +++ b/server/lib/events/s2storage_test.go @@ -64,3 +64,28 @@ func TestS2StorageController_EmptyStreamDoesNotStart(t *testing.T) { assert.False(t, c.Running()) assert.False(t, c.EverStarted()) } + +func TestS2StorageController_StopWithoutStart(t *testing.T) { + c, _ := newTestController(t, "test-stream") + + require.NoError(t, stopController(t, c)) + assert.False(t, c.EverStarted()) +} + +// TestS2StorageController_EverStartedSurvivesStop covers the state a caller +// reads to decide whether anything could have been persisted: Running goes back +// to false at shutdown, EverStarted does not. Restarting is not allowed either, +// since the append session binds a stream for its lifetime. +func TestS2StorageController_EverStartedSurvivesStop(t *testing.T) { + c, resolved := newTestController(t, "test-stream") + + require.NoError(t, c.Start(context.Background())) + require.NoError(t, stopController(t, c)) + + assert.False(t, c.Running()) + assert.True(t, c.EverStarted()) + + require.NoError(t, c.Start(context.Background())) + assert.False(t, c.Running(), "a stopped controller must not reopen the sink") + assert.Equal(t, int32(1), resolved.Load()) +} From e57b7acc57837495d07780183d128d06ac84c16f Mon Sep 17 00:00:00 2001 From: archandatta <35818003+archandatta@users.noreply.github.com> Date: Fri, 18 Sep 2026 12:21:26 +0000 Subject: [PATCH 4/7] Cover S2 controller start edge cases --- server/lib/events/s2storage.go | 5 ++- server/lib/events/s2storage_test.go | 57 +++++++++++++++++++++++++++++ 2 files changed, 61 insertions(+), 1 deletion(-) diff --git a/server/lib/events/s2storage.go b/server/lib/events/s2storage.go index c0e6c1b67..9fada235a 100644 --- a/server/lib/events/s2storage.go +++ b/server/lib/events/s2storage.go @@ -264,8 +264,11 @@ func (c *S2StorageController) Start(parent context.Context) error { if c.everStarted { return nil } + if c.basin == "" || c.token == "" { + return nil + } stream := c.streamFn() - if c.basin == "" || c.token == "" || stream == "" { + if stream == "" { return nil } c.log.Info("S2 storage enabled", "basin", c.basin, "stream", stream) diff --git a/server/lib/events/s2storage_test.go b/server/lib/events/s2storage_test.go index ab4a35d96..93d07b258 100644 --- a/server/lib/events/s2storage_test.go +++ b/server/lib/events/s2storage_test.go @@ -3,6 +3,7 @@ package events import ( "context" "log/slog" + "sync" "sync/atomic" "testing" "time" @@ -56,6 +57,49 @@ func TestS2StorageController_StartIsIdempotent(t *testing.T) { require.NoError(t, stopController(t, c)) } +func TestS2StorageController_ConcurrentStartOpensOneWriter(t *testing.T) { + c, resolved := newTestController(t, "test-stream") + + var wg sync.WaitGroup + errs := make(chan error, 16) + for range 16 { + wg.Add(1) + go func() { + defer wg.Done() + errs <- c.Start(context.Background()) + }() + } + wg.Wait() + close(errs) + + for err := range errs { + require.NoError(t, err) + } + assert.True(t, c.Running()) + assert.True(t, c.EverStarted()) + assert.Equal(t, int32(1), resolved.Load()) + + require.NoError(t, stopController(t, c)) +} + +func TestS2StorageController_StartFailureRollsBack(t *testing.T) { + c, resolved := newTestController(t, "test-stream") + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + require.ErrorIs(t, c.Start(ctx), context.Canceled) + assert.False(t, c.Running()) + assert.False(t, c.EverStarted()) + assert.Equal(t, int32(1), resolved.Load()) + + require.NoError(t, c.Start(context.Background())) + assert.True(t, c.Running()) + assert.True(t, c.EverStarted()) + assert.Equal(t, int32(2), resolved.Load()) + + require.NoError(t, stopController(t, c)) +} + func TestS2StorageController_EmptyStreamDoesNotStart(t *testing.T) { c, resolved := newTestController(t, "") @@ -65,6 +109,19 @@ func TestS2StorageController_EmptyStreamDoesNotStart(t *testing.T) { assert.False(t, c.EverStarted()) } +func TestS2StorageController_MissingCredentialsDoesNotResolveStream(t *testing.T) { + var resolved atomic.Int32 + c := NewS2StorageController(newTestStream(t, 64), "", "", func() string { + resolved.Add(1) + return "test-stream" + }, S2Config{}, slog.Default()) + + require.NoError(t, c.Start(context.Background())) + assert.Zero(t, resolved.Load()) + assert.False(t, c.Running()) + assert.False(t, c.EverStarted()) +} + func TestS2StorageController_StopWithoutStart(t *testing.T) { c, _ := newTestController(t, "test-stream") From bede878fde236e7b66aec7433332275a2f14addc Mon Sep 17 00:00:00 2001 From: archandatta <35818003+archandatta@users.noreply.github.com> Date: Fri, 18 Sep 2026 12:49:13 +0000 Subject: [PATCH 5/7] Trim comments that restate the code --- server/lib/events/s2storage.go | 25 ++++++------------------- server/lib/events/s2storage_test.go | 10 ---------- 2 files changed, 6 insertions(+), 29 deletions(-) diff --git a/server/lib/events/s2storage.go b/server/lib/events/s2storage.go index 9fada235a..5166be9d9 100644 --- a/server/lib/events/s2storage.go +++ b/server/lib/events/s2storage.go @@ -227,11 +227,8 @@ func (w *S2StorageWriter) Stop(ctx context.Context) error { return w.storage.Close(ctx) } -// S2StorageController owns the lifecycle of the S2 sink so the writer can be -// opened on demand rather than at boot. The append session binds one stream for -// its lifetime and the writer under it is single-use, so the controller opens at -// most one writer: once a writer has opened, Start is a no-op for the rest of -// the process. Safe for concurrent use. +// S2StorageController opens at most one writer because StorageWriter is +// single-use and an append session is bound to one stream. type S2StorageController struct { es *EventStream basin string @@ -246,18 +243,12 @@ type S2StorageController struct { everStarted bool } -// NewS2StorageController resolves the stream name through streamFn at Start -// rather than here: an instance holding for a fork identity carries the stream -// of the instance it was forked from until the handoff lands. +// NewS2StorageController resolves streamFn at Start because a fork learns its +// stream after construction. func NewS2StorageController(es *EventStream, basin, token string, streamFn func() string, cfg S2Config, log *slog.Logger) *S2StorageController { return &S2StorageController{es: es, basin: basin, token: token, streamFn: streamFn, cfg: cfg, log: log} } -// Start opens the sink, or is a no-op when a writer has already opened or the -// basin, token, or stream is unset. A failed open leaves the controller unstarted -// so a later identity can still open one. parent governs the read loop; the -// controller derives a cancelable child so Stop can halt the loop even when -// parent is still live. func (c *S2StorageController) Start(parent context.Context) error { c.mu.Lock() defer c.mu.Unlock() @@ -282,8 +273,6 @@ func (c *S2StorageController) Start(parent context.Context) error { return nil } -// Stop drains and shuts down a running writer, or is a no-op if none is running. -// ctx bounds shutdown time. func (c *S2StorageController) Stop(ctx context.Context) error { c.mu.Lock() defer c.mu.Unlock() @@ -296,16 +285,14 @@ func (c *S2StorageController) Stop(ctx context.Context) error { return err } -// Running reports whether the sink is currently forwarding events. func (c *S2StorageController) Running() bool { c.mu.Lock() defer c.mu.Unlock() return c.writer != nil } -// EverStarted reports whether a writer ever opened, including one already -// stopped. A caller that must guarantee nothing was persisted needs this rather -// than Running, which goes false again at shutdown. +// EverStarted remains true after Stop so callers can tell whether anything +// could have been persisted. func (c *S2StorageController) EverStarted() bool { c.mu.Lock() defer c.mu.Unlock() diff --git a/server/lib/events/s2storage_test.go b/server/lib/events/s2storage_test.go index 93d07b258..71683fd82 100644 --- a/server/lib/events/s2storage_test.go +++ b/server/lib/events/s2storage_test.go @@ -12,9 +12,6 @@ import ( "github.com/stretchr/testify/require" ) -// newTestController builds a controller whose streamFn returns stream and -// records how many times it was called, so a test can assert the name is -// resolved once, at Start. func newTestController(t *testing.T, stream string) (*S2StorageController, *atomic.Int32) { t.Helper() var resolved atomic.Int32 @@ -32,9 +29,6 @@ func stopController(t *testing.T, c *S2StorageController) error { return c.Stop(ctx) } -// TestS2StorageController_OpensNothingUntilStart is the guarantee the whole -// export-only mode rests on: a controller that is never started resolves no -// stream, so no append session is opened and nothing is persisted. func TestS2StorageController_OpensNothingUntilStart(t *testing.T) { c, resolved := newTestController(t, "test-stream") @@ -129,10 +123,6 @@ func TestS2StorageController_StopWithoutStart(t *testing.T) { assert.False(t, c.EverStarted()) } -// TestS2StorageController_EverStartedSurvivesStop covers the state a caller -// reads to decide whether anything could have been persisted: Running goes back -// to false at shutdown, EverStarted does not. Restarting is not allowed either, -// since the append session binds a stream for its lifetime. func TestS2StorageController_EverStartedSurvivesStop(t *testing.T) { c, resolved := newTestController(t, "test-stream") From 24ee4eb38691d3f2078feb87b124a18d0f0f96cd Mon Sep 17 00:00:00 2001 From: archandatta <35818003+archandatta@users.noreply.github.com> Date: Fri, 18 Sep 2026 19:38:39 +0000 Subject: [PATCH 6/7] Make S2 controller lifecycle nonblocking --- server/lib/events/s2storage.go | 72 ++++++++++++++--- server/lib/events/s2storage_test.go | 117 ++++++++++++++++++++++++++++ 2 files changed, 180 insertions(+), 9 deletions(-) diff --git a/server/lib/events/s2storage.go b/server/lib/events/s2storage.go index 5166be9d9..5610c6c6b 100644 --- a/server/lib/events/s2storage.go +++ b/server/lib/events/s2storage.go @@ -240,6 +240,8 @@ type S2StorageController struct { mu sync.Mutex writer *S2StorageWriter cancel context.CancelFunc + startDone chan struct{} + stopDone chan struct{} everStarted bool } @@ -251,13 +253,24 @@ func NewS2StorageController(es *EventStream, basin, token string, streamFn func( func (c *S2StorageController) Start(parent context.Context) error { c.mu.Lock() - defer c.mu.Unlock() - if c.everStarted { + if c.everStarted || c.startDone != nil { + c.mu.Unlock() return nil } if c.basin == "" || c.token == "" { + c.mu.Unlock() return nil } + startDone := make(chan struct{}) + c.startDone = startDone + c.mu.Unlock() + defer func() { + c.mu.Lock() + c.startDone = nil + close(startDone) + c.mu.Unlock() + }() + stream := c.streamFn() if stream == "" { return nil @@ -269,20 +282,61 @@ func (c *S2StorageController) Start(parent context.Context) error { cancel() return err } + c.mu.Lock() c.writer, c.cancel, c.everStarted = w, cancel, true + c.mu.Unlock() return nil } func (c *S2StorageController) Stop(ctx context.Context) error { - c.mu.Lock() - defer c.mu.Unlock() - if c.writer == nil { + for { + c.mu.Lock() + if c.startDone != nil { + startDone := c.startDone + c.mu.Unlock() + if err := waitForS2ControllerOperation(ctx, startDone); err != nil { + return err + } + continue + } + if c.stopDone != nil { + stopDone := c.stopDone + c.mu.Unlock() + if err := waitForS2ControllerOperation(ctx, stopDone); err != nil { + return err + } + continue + } + if c.writer == nil { + c.mu.Unlock() + return nil + } + writer, cancel := c.writer, c.cancel + stopDone := make(chan struct{}) + c.stopDone = stopDone + c.mu.Unlock() + + cancel() + err := writer.Stop(ctx) + + c.mu.Lock() + if err == nil { + c.writer, c.cancel = nil, nil + } + c.stopDone = nil + close(stopDone) + c.mu.Unlock() + return err + } +} + +func waitForS2ControllerOperation(ctx context.Context, done <-chan struct{}) error { + select { + case <-done: return nil + case <-ctx.Done(): + return ctx.Err() } - c.cancel() - err := c.writer.Stop(ctx) - c.writer, c.cancel = nil, nil - return err } func (c *S2StorageController) Running() bool { diff --git a/server/lib/events/s2storage_test.go b/server/lib/events/s2storage_test.go index 71683fd82..40bfb5d2c 100644 --- a/server/lib/events/s2storage_test.go +++ b/server/lib/events/s2storage_test.go @@ -123,6 +123,123 @@ func TestS2StorageController_StopWithoutStart(t *testing.T) { assert.False(t, c.EverStarted()) } +func TestS2StorageController_StopTimeoutKeepsWriter(t *testing.T) { + c := &S2StorageController{ + writer: &S2StorageWriter{started: true, done: make(chan struct{})}, + cancel: func() {}, + everStarted: true, + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + require.ErrorIs(t, c.Stop(ctx), context.Canceled) + assert.True(t, c.Running()) + assert.True(t, c.EverStarted()) + require.ErrorIs(t, c.Stop(ctx), context.Canceled) +} + +func TestS2StorageController_StopHonorsContextDuringStart(t *testing.T) { + entered := make(chan struct{}) + release := make(chan struct{}) + var releaseOnce sync.Once + releaseStart := func() { releaseOnce.Do(func() { close(release) }) } + defer releaseStart() + c := NewS2StorageController(newTestStream(t, 64), "test-basin", "test-token", func() string { + close(entered) + <-release + return "test-stream" + }, S2Config{}, slog.Default()) + parent, cancelParent := context.WithCancel(context.Background()) + cancelParent() + + startDone := make(chan error, 1) + go func() { startDone <- c.Start(parent) }() + <-entered + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + stopDone := make(chan error, 1) + go func() { stopDone <- c.Stop(ctx) }() + select { + case err := <-stopDone: + require.ErrorIs(t, err, context.Canceled) + case <-time.After(time.Second): + t.Fatal("Stop did not honor its context while Start was in progress") + } + + releaseStart() + require.ErrorIs(t, <-startDone, context.Canceled) +} + +func TestS2StorageController_StateAvailableDuringStop(t *testing.T) { + stopEntered := make(chan struct{}) + var cancelOnce sync.Once + c := &S2StorageController{ + writer: &S2StorageWriter{started: true, done: make(chan struct{})}, + cancel: func() { + cancelOnce.Do(func() { close(stopEntered) }) + }, + everStarted: true, + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + stopDone := make(chan error, 1) + go func() { stopDone <- c.Stop(ctx) }() + <-stopEntered + + runningDone := make(chan bool, 1) + go func() { runningDone <- c.Running() }() + select { + case running := <-runningDone: + assert.True(t, running) + case <-time.After(time.Second): + t.Fatal("Running blocked while Stop was in progress") + } + + everStartedDone := make(chan bool, 1) + go func() { everStartedDone <- c.EverStarted() }() + select { + case everStarted := <-everStartedDone: + assert.True(t, everStarted) + case <-time.After(time.Second): + t.Fatal("EverStarted blocked while Stop was in progress") + } + + cancel() + require.ErrorIs(t, <-stopDone, context.Canceled) +} + +func TestS2StorageController_StopHonorsContextDuringStop(t *testing.T) { + stopEntered := make(chan struct{}) + var cancelOnce sync.Once + c := &S2StorageController{ + writer: &S2StorageWriter{started: true, done: make(chan struct{})}, + cancel: func() { + cancelOnce.Do(func() { close(stopEntered) }) + }, + everStarted: true, + } + firstCtx, cancelFirst := context.WithCancel(context.Background()) + defer cancelFirst() + firstStopDone := make(chan error, 1) + go func() { firstStopDone <- c.Stop(firstCtx) }() + <-stopEntered + + secondCtx, cancelSecond := context.WithCancel(context.Background()) + cancelSecond() + secondStopDone := make(chan error, 1) + go func() { secondStopDone <- c.Stop(secondCtx) }() + select { + case err := <-secondStopDone: + require.ErrorIs(t, err, context.Canceled) + case <-time.After(time.Second): + t.Fatal("Stop did not honor its context while another Stop was in progress") + } + + cancelFirst() + require.ErrorIs(t, <-firstStopDone, context.Canceled) +} + func TestS2StorageController_EverStartedSurvivesStop(t *testing.T) { c, resolved := newTestController(t, "test-stream") From 4e921bbebf7ea6d2bee547abf4417b10f6f9a509 Mon Sep 17 00:00:00 2001 From: archandatta <35818003+archandatta@users.noreply.github.com> Date: Fri, 18 Sep 2026 19:39:25 +0000 Subject: [PATCH 7/7] Log S2 enablement after startup --- server/lib/events/s2storage.go | 2 +- server/lib/events/s2storage_test.go | 25 +++++++++++++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/server/lib/events/s2storage.go b/server/lib/events/s2storage.go index 5610c6c6b..e99cdb398 100644 --- a/server/lib/events/s2storage.go +++ b/server/lib/events/s2storage.go @@ -275,13 +275,13 @@ func (c *S2StorageController) Start(parent context.Context) error { if stream == "" { return nil } - c.log.Info("S2 storage enabled", "basin", c.basin, "stream", stream) runCtx, cancel := context.WithCancel(parent) w := NewS2StorageWriter(c.es, c.basin, c.token, stream, c.cfg, c.log) if err := w.Start(runCtx); err != nil { cancel() return err } + c.log.Info("S2 storage enabled", "basin", c.basin, "stream", stream) c.mu.Lock() c.writer, c.cancel, c.everStarted = w, cancel, true c.mu.Unlock() diff --git a/server/lib/events/s2storage_test.go b/server/lib/events/s2storage_test.go index 40bfb5d2c..e3eae4ee4 100644 --- a/server/lib/events/s2storage_test.go +++ b/server/lib/events/s2storage_test.go @@ -1,6 +1,7 @@ package events import ( + "bytes" "context" "log/slog" "sync" @@ -94,6 +95,30 @@ func TestS2StorageController_StartFailureRollsBack(t *testing.T) { require.NoError(t, stopController(t, c)) } +func TestS2StorageController_FailedStartDoesNotLogEnabled(t *testing.T) { + var logs bytes.Buffer + c := NewS2StorageController(newTestStream(t, 64), "test-basin", "test-token", func() string { + return "test-stream" + }, S2Config{}, slog.New(slog.NewTextHandler(&logs, nil))) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + require.ErrorIs(t, c.Start(ctx), context.Canceled) + assert.NotContains(t, logs.String(), "S2 storage enabled") +} + +func TestS2StorageController_SuccessfulStartLogsEnabled(t *testing.T) { + var logs bytes.Buffer + c := NewS2StorageController(newTestStream(t, 64), "test-basin", "test-token", func() string { + return "test-stream" + }, S2Config{}, slog.New(slog.NewTextHandler(&logs, nil))) + + require.NoError(t, c.Start(context.Background())) + assert.Contains(t, logs.String(), "S2 storage enabled") + + require.NoError(t, stopController(t, c)) +} + func TestS2StorageController_EmptyStreamDoesNotStart(t *testing.T) { c, resolved := newTestController(t, "")