Update trickle local publisher streaming behavior#3990
Conversation
📝 WalkthroughWalkthroughChangesTrickle streaming and lifecycle controls
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Publisher
participant TrickleServer
participant Segment
participant Subscriber
Publisher->>TrickleServer: POST with Lp-Trickle-Reset
TrickleServer->>Segment: close prior segments
Segment-->>Subscriber: unblock waiting read
TrickleServer->>Segment: write reset payload at next index
Segment-->>Subscriber: return reset payload and sequence headers
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
trickle/trickle_server.go (2)
690-699: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftReject writes after a segment is closed.
Reset closes prior segments, but in-flight publishers can subsequently append through
writeData. Those late bytes become visible to future readers, mixing pre-reset and post-reset data in a completed segment—the new test explicitly demonstrates this path. MakewriteDatareturn an error whenclosedand propagate it through local and HTTP publishers.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@trickle/trickle_server.go` around lines 690 - 699, Update Segment.writeData to return an error when the segment is closed, checking closed while holding segment.mutex before appending or broadcasting. Propagate this error through the local and HTTP publisher paths so late writes are rejected rather than exposed after reset.
466-490: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winCentralize stream-head advancement as a monotonic operation.
The duplicated assignments allow stale writers to rewind the head, while local zero-byte writes skip advancement entirely.
trickle/trickle_server.go#L466-L490: replace both direct assignments with a monotonic head-advance helper.trickle/local_publisher.go#L63-L68: use that helper on the first payload and when completing an empty segment.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@trickle/trickle_server.go` around lines 466 - 490, The stream head can be rewound by direct assignments, and zero-byte local segments do not advance it. In trickle/trickle_server.go lines 466-490, replace both direct s.nextWrite assignments with the shared monotonic head-advance helper; in trickle/local_publisher.go lines 63-68, use the same helper for the first payload and when completing an empty segment. Locate the affected logic via s.nextWrite and the local publisher’s segment completion handling.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@trickle/trickle_server.go`:
- Around line 345-351: Update writeHookError so non-RequestError failures are
logged server-side and returned with a generic HTTP 500 message instead of
exposing err.Error() to the client. Preserve the existing RequestError response
behavior and status handling unchanged.
- Around line 41-43: Update the publish-triggered stream creation path for POST
/{stream}/{idx} to invoke the BeforeCreate hook before autocreating a missing
stream, matching the existing HTTP channel creation path and propagating any
RequestError. Add a test covering a blocked publish attempt that verifies the
stream is not created.
In `@trickle/trickle_test.go`:
- Around line 556-559: Update the test goroutines around the publisher write and
subscriber read to send their returned errors and data through result channels
instead of discarding them. Replace unbounded receives at the affected sections
with timed select statements, and assert the publisher error, subscriber read
error, and received data while preserving the existing synchronization behavior.
- Around line 638-639: Update the test flow around sub.Read() to check and
handle err before accessing resp or registering the resp.Body.Close defer;
preserve the existing response-body cleanup only after a successful read.
---
Outside diff comments:
In `@trickle/trickle_server.go`:
- Around line 690-699: Update Segment.writeData to return an error when the
segment is closed, checking closed while holding segment.mutex before appending
or broadcasting. Propagate this error through the local and HTTP publisher paths
so late writes are rejected rather than exposed after reset.
- Around line 466-490: The stream head can be rewound by direct assignments, and
zero-byte local segments do not advance it. In trickle/trickle_server.go lines
466-490, replace both direct s.nextWrite assignments with the shared monotonic
head-advance helper; in trickle/local_publisher.go lines 63-68, use the same
helper for the first payload and when completing an empty segment. Locate the
affected logic via s.nextWrite and the local publisher’s segment completion
handling.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: e71199de-a585-4270-8dbe-1b34abc27e94
📒 Files selected for processing (6)
trickle/README.mdtrickle/local_publisher.gotrickle/segment_buffer.gotrickle/segment_buffer_test.gotrickle/trickle_server.gotrickle/trickle_test.go
| // BeforeCreate runs before HTTP channel creation. | ||
| // Return RequestError for expected client/policy failures. | ||
| BeforeCreate func(r *http.Request, streamName string) error |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
ast-grep outline trickle/trickle_server.go --items all \
--match 'handlePost|getOrCreateStream|BeforeCreate' --view expanded
rg -n -C3 '\bBeforeCreate\b|\bgetOrCreateStream\s*\(' trickleRepository: livepeer/go-livepeer
Length of output: 4769
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '180,250p' trickle/trickle_server.go
printf '\n---\n'
sed -n '320,460p' trickle/trickle_server.go
printf '\n---\n'
sed -n '200,330p' trickle/trickle_test.goRepository: livepeer/go-livepeer
Length of output: 9166
Apply BeforeCreate to publish-triggered creation too. POST /{stream}/{idx} can autocreate a missing stream, but that path skips BeforeCreate, so a publisher can bypass creation policy by posting directly to a segment. Route both creation paths through the hook and add a blocked-publish test.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@trickle/trickle_server.go` around lines 41 - 43, Update the publish-triggered
stream creation path for POST /{stream}/{idx} to invoke the BeforeCreate hook
before autocreating a missing stream, matching the existing HTTP channel
creation path and propagating any RequestError. Add a test covering a blocked
publish attempt that verifies the stream is not created.
| func writeHookError(w http.ResponseWriter, err error) { | ||
| var requestErr *RequestError | ||
| if errors.As(err, &requestErr) { | ||
| http.Error(w, requestErr.Error(), requestErr.httpStatus()) | ||
| return | ||
| } | ||
| http.Error(w, err.Error(), http.StatusInternalServerError) |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Do not expose generic hook errors to clients.
err.Error() may contain internal policy, database, or identity details. Log it server-side and return a generic 500 body.
Proposed fix
if errors.As(err, &requestErr) {
http.Error(w, requestErr.Error(), requestErr.httpStatus())
return
}
- http.Error(w, err.Error(), http.StatusInternalServerError)
+ slog.Error("HTTP lifecycle hook failed", "err", err)
+ http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| func writeHookError(w http.ResponseWriter, err error) { | |
| var requestErr *RequestError | |
| if errors.As(err, &requestErr) { | |
| http.Error(w, requestErr.Error(), requestErr.httpStatus()) | |
| return | |
| } | |
| http.Error(w, err.Error(), http.StatusInternalServerError) | |
| func writeHookError(w http.ResponseWriter, err error) { | |
| var requestErr *RequestError | |
| if errors.As(err, &requestErr) { | |
| http.Error(w, requestErr.Error(), requestErr.httpStatus()) | |
| return | |
| } | |
| slog.Error("HTTP lifecycle hook failed", "err", err) | |
| http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@trickle/trickle_server.go` around lines 345 - 351, Update writeHookError so
non-RequestError failures are logged server-side and returned with a generic
HTTP 500 message instead of exposing err.Error() to the client. Preserve the
existing RequestError response behavior and status handling unchanged.
| writeDone := make(chan struct{}) | ||
| go func() { | ||
| defer close(writeDone) | ||
| _ = lp.Write(r0) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Propagate goroutine errors and bound channel waits.
The publisher error is discarded, the subscriber read error is ignored, and both receives can hang until the suite timeout. Return results through channels, assert the errors/data, and use a timed select.
Also applies to: 575-593, 616-617
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@trickle/trickle_test.go` around lines 556 - 559, Update the test goroutines
around the publisher write and subscriber read to send their returned errors and
data through result channels instead of discarding them. Replace unbounded
receives at the affected sections with timed select statements, and assert the
publisher error, subscriber read error, and received data while preserving the
existing synchronization behavior.
| resp, err := sub.Read() | ||
| defer resp.Body.Close() |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Check err before dereferencing resp.
A failed sub.Read() currently panics in the deferred call and hides the actual failure.
resp, err := sub.Read()
+ require.NoError(err)
defer resp.Body.Close()📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| resp, err := sub.Read() | |
| defer resp.Body.Close() | |
| resp, err := sub.Read() | |
| require.NoError(err) | |
| defer resp.Body.Close() |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@trickle/trickle_test.go` around lines 638 - 639, Update the test flow around
sub.Read() to check and handle err before accessing resp or registering the
resp.Body.Close defer; preserve the existing response-body cleanup only after a
successful read.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #3990 +/- ##
===================================================
+ Coverage 33.28594% 33.53732% +0.25138%
===================================================
Files 171 172 +1
Lines 42204 42323 +119
===================================================
+ Hits 14048 14194 +146
+ Misses 27101 27063 -38
- Partials 1055 1066 +11
... and 5 files with indirect coverage changes Continue to review full report in Codecov by Harness.
🚀 New features to boost your workflow:
|
rickstaa
left a comment
rickstaa
left a comment
There was a problem hiding this comment.
@j0sh, confirmed that this is byte-for-byte identical to the code I reviewed and tested in https://github.com/livepeer/go-livepeer/pull/3938/files. Good to merge.
My only comment on the combined PR was that the third argument to getOrCreateStream is a bit ambiguous: https://github.com/livepeer/go-livepeer/pull/3938/files#r3505950363. If you agree you could include this in this PR or a follow up PR.
|
Renamed |
2 participants
Summary
Tests
The test runs used an unstaged local replace: github.com/livepeer/lpms => /Users/josh/lpms.
Summary by CodeRabbit
New Features
Bug Fixes
Documentation