ai/live: Add limited retries for segment publish

[j0sh]

Only retry if the error occurs before sending any data, and if the next segment hasn't arrived yet.

Depends on #3308

[codecov]

Codecov Report

Attention: Patch coverage is 0% with 86 lines in your changes missing coverage. Please review.

Project coverage is 33.86599%. Comparing base (a842a0d) to head (2ea2080).
Report is 2 commits behind head on master.

Files with missing lines	Patch %	Lines
trickle/trickle_publisher.go	0.00000%	48 Missing ⚠️
server/ai_live_video.go	0.00000%	33 Missing ⚠️
trickle/trickle_server.go	0.00000%	5 Missing ⚠️

Additional details and impacted files

@@                 Coverage Diff                 @@
##              master       #3315         +/-   ##
===================================================
- Coverage   33.92247%   33.86599%   -0.05648%     
===================================================
  Files            141         141                 
  Lines          37173       37235         +62     
===================================================
  Hits           12610       12610                 
- Misses         23843       23905         +62     
  Partials         720         720                 

Files with missing lines	Coverage Δ
trickle/trickle_server.go	0.00000% <0.00000%> (ø)
server/ai_live_video.go	0.00000% <0.00000%> (ø)
trickle/trickle_publisher.go	0.00000% <0.00000%> (ø)

---

Continue to review full report in Codecov by Sentry.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 3ac5972...2ea2080. Read the comment docs.

Files with missing lines	Coverage Δ
trickle/trickle_server.go	0.00000% <0.00000%> (ø)
server/ai_live_video.go	0.00000% <0.00000%> (ø)
trickle/trickle_publisher.go	0.00000% <0.00000%> (ø)

[victorges]

lgtm

[victorges]

Hmm this seems maybe not 100% guaranteed, since it could happen that something was read from the reader and then failed to be sent to the writer (and we got 0 on L89).

I think there's no really easy option here apart from maybe an additional layer to check if the reader was used at some point, like a wrapped reader that flags if it was read. A built-in approach could be using a io.TeeReader to read into a buffer as we read from the input (which would also allow us to retry when some bytes have been read).

Anyway, if you think this is too much of a corner case, this LGTM as it's much simpler. But I think that conceptually it could still have this issue 🤔

[j0sh]

since it could happen that something was read from the reader and then failed to be sent to the writer (and we got 0 on L89).

That is basically why we call Clone() (which resets the read head for the clone) so when we try to read it during the next iteration, it is starting from the beginning, regardless of the actual read head. Note that this isn't a "standard" IO reader but rather something specialized that allows for clean resets, nonblocking reads/writes, multiple readers, etc, without having to go through hoops like TeeReader.

[victorges]

Ohh that's cool! Didn't know we already had that custom impl. Pretty neat :)