perf: fix spin lock in consume stream call

[Devin-Yeung]

Summary by CodeRabbit

• New Features

  • Added a blocking wait for new log entries to replace retry-based polling, improving streaming efficiency and reducing CPU use.
  • Stream handling now respects cancellations and timeouts for cleaner shutdowns.

• Bug Fixes

  • Prevents busy-spin behavior when consumers request out-of-range offsets.

• Tests

  • Added tests covering blocking-wait behavior, cancellation handling, and non-busy streaming.

[coderabbitai]

📝 Walkthrough

Walkthrough

Adds a notify channel and WaitForAppend(ctx) to the log, updates server ConsumeStream to block on WaitForAppend when offsets are out of range instead of busy-waiting, and adds tests validating wait and non-busy behavior.

Changes

Cohort / File(s)	Summary
Log Notification Mechanism internal/log/log.go	Adds a notify channel to Log, initializes it in NewLog, broadcasts on append by closing/replacing the channel, and implements (*Log) WaitForAppend(ctx context.Context) error.
Log Tests internal/log/log_test.go	Adds testWaitForAppend and testWaitForAppendContextCanceled and registers them in the test suite to validate blocking behavior and context cancellation.
Server Integration internal/server/server.go	Adds WaitForAppend(ctx context.Context) error to the CommitLog interface and refactors ConsumeStream to call WaitForAppend on ErrOffsetOutOfRange, replacing the prior busy-retry loop.
Server busy-loop test internal/server/server_busy_loop_test.go	Adds a test ensuring ConsumeStream does not busy-spin when offset is out-of-range by counting Read calls and asserting limited retries/backoff.

Sequence Diagram

sequenceDiagram
    participant Client
    participant ConsumeStream
    participant CommitLog
    participant Context

    Client->>ConsumeStream: Open stream / request Consume(offset)
    ConsumeStream->>CommitLog: Consume(offset)
    CommitLog-->>ConsumeStream: ErrOffsetOutOfRange

    ConsumeStream->>CommitLog: WaitForAppend(ctx)
    Note over CommitLog: Blocks waiting on notify channel

    par Append occurs elsewhere
        CommitLog->>CommitLog: Append(record) — close & replace notify
        CommitLog-->>ConsumeStream: notify (channel closed) — wake waiter
    end

    ConsumeStream->>CommitLog: Consume(offset)
    CommitLog-->>ConsumeStream: Record
    ConsumeStream->>Client: Send record

Loading

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~25 minutes

Possibly related PRs

• feat: a high-level persistent log api #1: Prior PR that added the base Log type and Append/Read APIs; directly related as this change extends Log with notify/WaitForAppend.
• feat: expose a rpc server for log service #2: Prior PR that modified internal/server/server.go; related because this change updates ConsumeStream and the CommitLog interface to use WaitForAppend.

Poem

🐰 I sat and twitched beside the log today,
No frantic hopping, no mad dash away.
A gentle bell of closure—soft and bright—
It told me "Append!" and set my heart alight.
Hooray for channels; now I nap till night. 🥕✨

🚥 Pre-merge checks | ✅ 2 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 30.00% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (2 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title accurately describes the main change: fixing a spin lock (busy-wait pattern) in the consume stream call by replacing it with a blocking wait mechanism via WaitForAppend.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing touches

• 📝 Generate docstrings

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch consume-stream-perf

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coveralls]

coverage: 39.644% (+1.1%) from 38.552%
when pulling cef6222 on consume-stream-perf
into 3a66e69 on master.

[coderabbitai]

Actionable comments posted: 1

🤖 Fix all issues with AI agents

In `@internal/server/server.go`:
- Around line 141-155: The Consume/WaitForAppend sequence can miss an append
because WaitForAppend may start waiting after the append happened; fix by
capturing the commit-log notify channel before calling s.Consume so the wait
corresponds to the checked state: add a Notify() <-chan struct{} method to the
CommitLog interface and implement it on Log (returning the l.notify channel
under lock), then in server.go call ch := s.CommitLog.Notify() before record,
err := s.Consume(...); on api.ErrOffsetOutOfRange do a select that waits on
either ctx.Done() or ch (instead of calling WaitForAppend), looping on notify
wakeups until Consume succeeds. Ensure Notify() is documented and used where
WaitForAppend previously blocked to avoid the race.