diff --git a/pkg/director/s3_upload.go b/pkg/director/s3_upload.go index 7c929666..4e158e0e 100644 --- a/pkg/director/s3_upload.go +++ b/pkg/director/s3_upload.go @@ -68,10 +68,17 @@ func (ss *StreamSession) shouldRecordLivestream(ctx context.Context, repoDID str func (ss *StreamSession) maybeStartS3Upload(ctx context.Context, repoDID string) { if !ss.cli.S3Configured() { + // Debug: the no-S3 dev default, so one line per stream start would be + // pure noise there — but it's flippable at runtime when an operator is + // asking "why isn't this node recording anything?". + log.Debug(ctx, "live recording skipped: S3 not configured", "repoDID", repoDID) return } if !ss.shouldRecordLivestream(ctx, repoDID) { - log.Debug(ctx, "live recording disabled for streamer (not in VOD beta or recording not enabled)", "repoDID", repoDID) + // Info, not Debug: this fires once per stream session, and it's the + // answer to "why does this streamer have no recording?" — a question + // that once cost a production debugging session at Debug level. + log.Log(ctx, "live recording disabled for streamer (not in VOD beta, or recording turned off in settings)", "repoDID", repoDID) return } cfg := ss.cli.S3Config() @@ -84,10 +91,12 @@ func (ss *StreamSession) maybeStartS3Upload(ctx context.Context, repoDID string) // the current stream everywhere (notification blast, idle finalize), so we // do the same here; NewSegment refreshes it once the stream's own record is // indexed, in case a prior stream was momentarily still "latest". - if ls, err := ss.mod.GetLatestLivestreamForRepo(repoDID); err == nil && ls != nil { + if ls, err := ss.mod.GetLatestLivestreamForRepo(repoDID); err != nil { + log.Warn(ctx, "live recording: failed to resolve initial livestream URI; first object starts untagged", "error", err, "repoDID", repoDID) + } else if ls != nil { ss.s3Uploader.SetLivestreamURI(ls.URI) } - log.Log(ctx, "S3 upload enabled", "bucket", ss.cli.S3Bucket, "endpoint", ss.cli.S3Endpoint) + log.Log(ctx, "S3 upload enabled", "bucket", ss.cli.S3Bucket, "endpoint", ss.cli.S3Endpoint, "repoDID", repoDID) } func (ss *StreamSession) s3Upload(ctx context.Context, notif *media.NewSegmentNotification) { diff --git a/pkg/s3/s3.go b/pkg/s3/s3.go index 18641624..2351f97c 100644 --- a/pkg/s3/s3.go +++ b/pkg/s3/s3.go @@ -206,10 +206,13 @@ func (u *S3Uploader) Cutover(ctx context.Context) error { } // Close signals that no more segments will be added, waits for all in-flight -// uploads to complete, and returns any error. It is idempotent: repeated calls -// return the same result without re-closing the channel. The supplied ctx is -// unused for the wait (uploadLoop runs on its own context so it can flush the -// final object even after the session context is cancelled) but kept for API +// uploads to complete, and returns any error completing the final object. +// Mid-stream upload failures don't surface here — the upload loop recovers +// from those by abandoning the broken object (logged loudly at the time) and +// continuing with a fresh one. It is idempotent: repeated calls return the +// same result without re-closing the channel. The supplied ctx is unused for +// the wait (uploadLoop runs on its own context so it can flush the final +// object even after the session context is cancelled) but kept for API // symmetry. func (u *S3Uploader) Close(ctx context.Context) error { u.closeOnce.Do(func() { @@ -369,15 +372,47 @@ func (u *S3Uploader) uploadLoop(ctx context.Context) { return nil } - var err error - for err == nil { + // abandonCurrent is the failure recovery: abort the broken object (so the + // backend doesn't hold its parts) and drop its un-completed bytes, loudly. + // The next segment starts a fresh object, so one bad object costs a gap in + // the recording instead of wedging the uploader for the rest of the stream + // (the abandoned object's recorder row never completes, so finalize skips + // it). Before this existed, the first error killed the loop: segments + // backed up silently and the stream never recorded another byte. + abandonCurrent := func(reason error) { + if current == nil { + // Nothing in flight (e.g. CreateMultipartUpload itself failed); the + // segment is still dropped, so say so. + log.Error(ctx, "error in live-rec S3 upload; segment dropped", "error", reason) + return + } + log.Error(ctx, "abandoning live-rec S3 object; its bytes will be missing from the recording", + "key", current.key, + "uploadedBytes", current.totalSize, + "droppedBufferedBytes", len(current.buf), + "error", reason, + ) + if _, err := u.client.AbortMultipartUpload(ctx, &s3.AbortMultipartUploadInput{ + Bucket: aws.String(u.bucket), + Key: aws.String(current.key), + UploadId: aws.String(current.uploadID), + }); err != nil { + log.Error(ctx, "aborting abandoned S3 upload", "key", current.key, "error", err) + } + current = nil + } + + for { select { case cmd, ok := <-u.segCh: if !ok { - // No more segments; complete any in-progress upload. - err = completeUpload() + // No more segments; complete any in-progress upload. This is the + // one error that still surfaces through done/Close — there are no + // more segments coming to recover with. + err := completeUpload() if err != nil { err = fmt.Errorf("error completing upload: %w", err) + abandonCurrent(err) } u.done <- err return @@ -385,26 +420,26 @@ func (u *S3Uploader) uploadLoop(ctx context.Context) { if cmd.cutover { // Close out the current object so it's immediately finalize-able // (e.g. the livestream just ended). No-op if nothing is in flight. - if err = completeUpload(); err != nil { - err = fmt.Errorf("error completing upload on cutover: %w", err) - log.Error(ctx, "error completing upload on cutover", "error", err) + if err := completeUpload(); err != nil { + abandonCurrent(fmt.Errorf("error completing upload on cutover: %w", err)) } continue } log.Debug(ctx, "received segment for S3 upload", "size", len(cmd.seg)) - if err = handleSegment(cmd.seg); err != nil { - log.Error(ctx, "error handling segment", "error", err) + if err := handleSegment(cmd.seg); err != nil { + // The triggering segment is dropped along with the object: it may + // already be partially flushed into it, so it can't be salvaged. + abandonCurrent(fmt.Errorf("error handling segment: %w", err)) } case <-ctx.Done(): - err = completeUpload() + err := completeUpload() if err != nil { err = fmt.Errorf("error completing upload: %w", err) + abandonCurrent(err) } u.done <- err return } } - - u.done <- err } diff --git a/pkg/s3/uploader_test.go b/pkg/s3/uploader_test.go index e2531a9d..b344c599 100644 --- a/pkg/s3/uploader_test.go +++ b/pkg/s3/uploader_test.go @@ -14,11 +14,15 @@ import ( ) // fakeUploadAPI is an in-memory stand-in for the multipart subset of *s3.Client -// the upload loop drives. It hands back dummy upload IDs / etags and never errs. +// the upload loop drives. It hands back dummy upload IDs / etags; set +// failCompletes to make the first N CompleteMultipartUpload calls fail. type fakeUploadAPI struct { - mu sync.Mutex - creates int - partSizes []int + mu sync.Mutex + creates int + partSizes []int + failCompletes int + completes int + aborts int } func (f *fakeUploadAPI) CreateMultipartUpload(_ context.Context, _ *awss3.CreateMultipartUploadInput, _ ...func(*awss3.Options)) (*awss3.CreateMultipartUploadOutput, error) { @@ -41,10 +45,20 @@ func (f *fakeUploadAPI) UploadPart(_ context.Context, in *awss3.UploadPartInput, } func (f *fakeUploadAPI) CompleteMultipartUpload(_ context.Context, _ *awss3.CompleteMultipartUploadInput, _ ...func(*awss3.Options)) (*awss3.CompleteMultipartUploadOutput, error) { + f.mu.Lock() + defer f.mu.Unlock() + if f.failCompletes > 0 { + f.failCompletes-- + return nil, fmt.Errorf("InvalidPart: all non-trailing parts must have the same length") + } + f.completes++ return &awss3.CompleteMultipartUploadOutput{}, nil } func (f *fakeUploadAPI) AbortMultipartUpload(_ context.Context, _ *awss3.AbortMultipartUploadInput, _ ...func(*awss3.Options)) (*awss3.AbortMultipartUploadOutput, error) { + f.mu.Lock() + f.aborts++ + f.mu.Unlock() return &awss3.AbortMultipartUploadOutput{}, nil } @@ -189,6 +203,36 @@ func TestS3UploaderUniformParts(t *testing.T) { require.Equal(t, total, got, "flushed parts must cover every byte exactly once") } +// TestS3UploaderRecoversFromCompleteFailure proves a failed +// CompleteMultipartUpload doesn't wedge the uploader: the broken object is +// aborted and abandoned, and the next segment starts a fresh object that +// uploads normally. Before this behavior existed, the first error killed the +// upload loop and the rest of the stream was silently never recorded — which +// is exactly how R2's InvalidPart rejection presented in production. +func TestS3UploaderRecoversFromCompleteFailure(t *testing.T) { + fc := &fakeUploadAPI{failCompletes: 1} + rec := &fakeRecorder{} + u := newS3Uploader(fc, "bucket", "did:plc:test", "did:plc:test/", time.Hour, rec) + + ctx := context.Background() + seg := make([]byte, 1024) + + require.NoError(t, u.AddSegment(ctx, seg)) + waitForStarts(t, rec, 1) // object 1 + require.NoError(t, u.Cutover(ctx)) // complete fails -> object 1 abandoned+aborted + + require.NoError(t, u.AddSegment(ctx, seg)) + waitForStarts(t, rec, 2) // loop survived: object 2 started + + require.NoError(t, u.Close(ctx), "mid-stream failure must not surface at Close; the final object completed fine") + + fc.mu.Lock() + defer fc.mu.Unlock() + require.Equal(t, 2, fc.creates, "a fresh object must start after the failure") + require.Equal(t, 1, fc.aborts, "the broken object must be aborted, not leaked") + require.Equal(t, 1, fc.completes, "the post-failure object must complete") +} + // TestS3UploaderCloseIdempotent exercises the lifecycle fix that re-enabled // live S3 upload: Close must be safe to call repeatedly and concurrently (it // was a plain close(segCh) before, which panicked on the second call), and