From 1cd17cd0fb0c17f4a7fe222d83f3acd5de3e05bf Mon Sep 17 00:00:00 2001 From: Eli Mallon Date: Sun, 7 Jun 2026 15:27:27 -0700 Subject: [PATCH] s3: roll over the live upload object when the livestream changes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A single continuous ingest moves through several place.stream.livestream records — each "update livestream" mints a NEW record (chapter markers), baked into subsequent segments' manifests. The S3 uploader stamped each object with one livestream URI at object-start, so an object could straddle a chapter change; finalize(newChapter) then matched no object and returned "no segments found". Cut a fresh object the moment the livestream URI changes (alongside the existing time-based cutover), so every object belongs to exactly one livestream. finalize(URI) is unchanged and now selects exactly that chapter's objects. This also composes with multi-node: a livestream that spans two nodes' StreamSessions tags both nodes' objects with the same shared URI into the shared bucket/statedb, so finalize coalesces them into one VOD (the reconnect's tfdt reset in the middle is the clean EXT-X-DISCONTINUITY case). Per-record VODs for now; a cross-record playlist can stitch a whole multi-chapter stream later. Also: object keys gain a per-uploader sequence number so a same-second rollover (now possible with per-chapter cutover) can't collide/overwrite; and the upload loop's S3 calls go through an injectable interface so the rollover is unit-tested with a fake client. Co-Authored-By: Claude Opus 4.8 --- pkg/s3/s3.go | 71 ++++++++++++++++++-------- pkg/s3/uploader_test.go | 109 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 158 insertions(+), 22 deletions(-) diff --git a/pkg/s3/s3.go b/pkg/s3/s3.go index 7d4c0f17..008ee1cc 100644 --- a/pkg/s3/s3.go +++ b/pkg/s3/s3.go @@ -35,6 +35,15 @@ type Recorder interface { RecordComplete(ctx context.Context, id string, parts int32, size int64) error } +// uploadAPI is the subset of *s3.Client the upload loop uses. Pulled out so +// tests can inject a fake; *s3.Client satisfies it. +type uploadAPI interface { + CreateMultipartUpload(context.Context, *s3.CreateMultipartUploadInput, ...func(*s3.Options)) (*s3.CreateMultipartUploadOutput, error) + UploadPart(context.Context, *s3.UploadPartInput, ...func(*s3.Options)) (*s3.UploadPartOutput, error) + CompleteMultipartUpload(context.Context, *s3.CompleteMultipartUploadInput, ...func(*s3.Options)) (*s3.CompleteMultipartUploadOutput, error) + AbortMultipartUpload(context.Context, *s3.AbortMultipartUploadInput, ...func(*s3.Options)) (*s3.AbortMultipartUploadOutput, error) +} + // S3Uploader manages streaming multipart uploads to an S3-compatible endpoint. // Bare canonical MUXL segments are fed via AddSegment and written verbatim — // no per-object init header — so the objects of one stream concatenate @@ -43,7 +52,7 @@ type Recorder interface { // what lets it "concat fearlessly". Every cutoverEvery, the current upload is // completed and a new one begins. type S3Uploader struct { - client *s3.Client + client uploadAPI bucket string cutoverEvery time.Duration keyPrefix string // e.g. "did:plc:abc123/" @@ -60,9 +69,13 @@ type S3Uploader struct { closed atomic.Bool } -// SetLivestreamURI records the livestream this stream's objects belong to. It -// may be called once the URI is resolved (it can be unknown when the uploader -// starts); subsequent multipart objects are stamped with it via RecordStart. +// SetLivestreamURI records the livestream this stream's objects belong to. A +// single continuous ingest can move through several place.stream.livestream +// records (each "update livestream" mints a new one — chapter markers), so when +// this changes the upload loop rolls over to a fresh object tagged with the new +// URI. That keeps every object within a single livestream, which is what lets +// finalize select exactly one livestream's objects (and, across nodes, coalesce +// them by the shared URI). It may be called before the URI is first resolved. func (u *S3Uploader) SetLivestreamURI(uri string) { u.mu.Lock() u.livestreamURI = uri @@ -79,14 +92,15 @@ func (u *S3Uploader) getLivestreamURI() string { const minPartSize = 5 * 1024 * 1024 type activeUpload struct { - key string - uploadID string - recordID string // set by Recorder.RecordStart, used for RecordComplete - parts []types.CompletedPart - partNum int32 - started time.Time - buf []byte // accumulates segments until we hit minPartSize - totalSize int64 // running total of bytes flushed across all parts + key string + uploadID string + recordID string // set by Recorder.RecordStart, used for RecordComplete + livestreamURI string // the livestream this object belongs to; a change rolls it over + parts []types.CompletedPart + partNum int32 + started time.Time + buf []byte // accumulates segments until we hit minPartSize + totalSize int64 // running total of bytes flushed across all parts } var DefaultCutoverEvery = 10 * time.Minute @@ -97,7 +111,6 @@ var DefaultCutoverEvery = 10 * time.Minute // nil to disable persistence. Starts the muxl Concatenator and a background // goroutine that reads processed segments and uploads them. func NewS3Uploader(cfg Config, userDID, keyPrefix string, cutoverEvery time.Duration, recorder Recorder) *S3Uploader { - ctx := context.Background() client := s3.New(s3.Options{ Region: cfg.Region, Credentials: credentials.NewStaticCredentialsProvider( @@ -108,12 +121,18 @@ func NewS3Uploader(cfg Config, userDID, keyPrefix string, cutoverEvery time.Dura BaseEndpoint: aws.String(cfg.Endpoint), UsePathStyle: true, }) + return newS3Uploader(client, cfg.Bucket, userDID, keyPrefix, cutoverEvery, recorder) +} + +// newS3Uploader is the client-injectable constructor behind NewS3Uploader; the +// fake-client tests use it directly. +func newS3Uploader(client uploadAPI, bucket, userDID, keyPrefix string, cutoverEvery time.Duration, recorder Recorder) *S3Uploader { if cutoverEvery == 0 { cutoverEvery = DefaultCutoverEvery } u := &S3Uploader{ client: client, - bucket: cfg.Bucket, + bucket: bucket, cutoverEvery: cutoverEvery, keyPrefix: keyPrefix, userDID: userDID, @@ -121,7 +140,7 @@ func NewS3Uploader(cfg Config, userDID, keyPrefix string, cutoverEvery time.Dura done: make(chan error, 1), recorder: recorder, } - go u.uploadLoop(ctx) + go u.uploadLoop(context.Background()) return u } @@ -168,10 +187,12 @@ func (u *S3Uploader) Close(ctx context.Context) error { func (u *S3Uploader) uploadLoop(ctx context.Context) { ctx = log.WithLogValues(ctx, "func", "s3.uploadLoop") var current *activeUpload + objSeq := 0 // disambiguates keys when two objects roll over within one second startUpload := func() error { + objSeq++ now := time.Now() - key := fmt.Sprintf("%s%s.m4s", u.keyPrefix, now.UTC().Format("2006-01-02T15-04-05")) + key := fmt.Sprintf("%s%s-%d.m4s", u.keyPrefix, now.UTC().Format("2006-01-02T15-04-05"), objSeq) resp, err := u.client.CreateMultipartUpload(ctx, &s3.CreateMultipartUploadInput{ Bucket: aws.String(u.bucket), @@ -182,13 +203,15 @@ func (u *S3Uploader) uploadLoop(ctx context.Context) { return fmt.Errorf("creating multipart upload for %s: %w", key, err) } + uri := u.getLivestreamURI() current = &activeUpload{ - key: key, - uploadID: *resp.UploadId, - started: now, + key: key, + uploadID: *resp.UploadId, + started: now, + livestreamURI: uri, } if u.recorder != nil { - id, recErr := u.recorder.RecordStart(ctx, u.userDID, u.bucket, key, u.getLivestreamURI(), now) + id, recErr := u.recorder.RecordStart(ctx, u.userDID, u.bucket, key, uri, now) if recErr != nil { log.Error(ctx, "recording S3 upload start", "key", key, "error", recErr) } @@ -268,8 +291,12 @@ func (u *S3Uploader) uploadLoop(ctx context.Context) { handleSegment := func(seg []byte) error { now := time.Now() - // Cut over if needed - if current != nil && now.Sub(current.started) >= u.cutoverEvery { + // Roll over to a new object when the current one has run for cutoverEvery, + // or when the livestream changed (a new place.stream.livestream "chapter" + // record). Cutting over on the livestream change keeps each object within + // a single livestream so finalize can select exactly one livestream's + // objects without one straddling two chapters. + if current != nil && (now.Sub(current.started) >= u.cutoverEvery || current.livestreamURI != u.getLivestreamURI()) { if err := completeUpload(); err != nil { return err } diff --git a/pkg/s3/uploader_test.go b/pkg/s3/uploader_test.go index 8324cd63..69a498e0 100644 --- a/pkg/s3/uploader_test.go +++ b/pkg/s3/uploader_test.go @@ -2,11 +2,120 @@ package s3 import ( "context" + "fmt" "sync" "testing" "time" + + "github.com/aws/aws-sdk-go-v2/aws" + awss3 "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/stretchr/testify/require" ) +// 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. +type fakeUploadAPI struct { + mu sync.Mutex + creates int +} + +func (f *fakeUploadAPI) CreateMultipartUpload(_ context.Context, _ *awss3.CreateMultipartUploadInput, _ ...func(*awss3.Options)) (*awss3.CreateMultipartUploadOutput, error) { + f.mu.Lock() + f.creates++ + n := f.creates + f.mu.Unlock() + return &awss3.CreateMultipartUploadOutput{UploadId: aws.String(fmt.Sprintf("up-%d", n))}, nil +} + +func (f *fakeUploadAPI) UploadPart(_ context.Context, _ *awss3.UploadPartInput, _ ...func(*awss3.Options)) (*awss3.UploadPartOutput, error) { + return &awss3.UploadPartOutput{ETag: aws.String("etag")}, nil +} + +func (f *fakeUploadAPI) CompleteMultipartUpload(_ context.Context, _ *awss3.CompleteMultipartUploadInput, _ ...func(*awss3.Options)) (*awss3.CompleteMultipartUploadOutput, error) { + return &awss3.CompleteMultipartUploadOutput{}, nil +} + +func (f *fakeUploadAPI) AbortMultipartUpload(_ context.Context, _ *awss3.AbortMultipartUploadInput, _ ...func(*awss3.Options)) (*awss3.AbortMultipartUploadOutput, error) { + return &awss3.AbortMultipartUploadOutput{}, nil +} + +// fakeRecorder captures the (key, livestreamURI) of every started object. +type fakeRecorder struct { + mu sync.Mutex + keys []string + uris []string + starts int +} + +func (r *fakeRecorder) RecordStart(_ context.Context, _, _, key, livestreamURI string, _ time.Time) (string, error) { + r.mu.Lock() + defer r.mu.Unlock() + r.keys = append(r.keys, key) + r.uris = append(r.uris, livestreamURI) + r.starts++ + return fmt.Sprintf("rec-%d", r.starts), nil +} + +func (r *fakeRecorder) RecordComplete(_ context.Context, _ string, _ int32, _ int64) error { + return nil +} + +func (r *fakeRecorder) count() int { r.mu.Lock(); defer r.mu.Unlock(); return r.starts } + +func (r *fakeRecorder) startURIs() []string { + r.mu.Lock() + defer r.mu.Unlock() + return append([]string(nil), r.uris...) +} + +func (r *fakeRecorder) startKeys() []string { + r.mu.Lock() + defer r.mu.Unlock() + return append([]string(nil), r.keys...) +} + +func waitForStarts(t *testing.T, rec *fakeRecorder, n int) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if rec.count() >= n { + return + } + time.Sleep(2 * time.Millisecond) + } + t.Fatalf("timed out waiting for %d RecordStart calls (got %d)", n, rec.count()) +} + +// TestS3UploaderCutoverOnLivestreamChange proves the uploader rolls over to a +// fresh object the moment the livestream URI changes (a new "chapter" record), +// so each object belongs to exactly one livestream — which is what lets finalize +// select one livestream's objects. cutoverEvery is set huge so ONLY the +// livestream change can trigger the rollover. +func TestS3UploaderCutoverOnLivestreamChange(t *testing.T) { + fc := &fakeUploadAPI{} + rec := &fakeRecorder{} + u := newS3Uploader(fc, "bucket", "did:plc:test", "did:plc:test/", time.Hour, rec) + + ctx := context.Background() + seg := make([]byte, 1024) // well under minPartSize: buffered until the object completes + + u.SetLivestreamURI("at://A") + require.NoError(t, u.AddSegment(ctx, seg)) + waitForStarts(t, rec, 1) // object 1, livestream A + + u.SetLivestreamURI("at://B") + require.NoError(t, u.AddSegment(ctx, seg)) + waitForStarts(t, rec, 2) // livestream changed -> object 2, livestream B + + require.NoError(t, u.Close(ctx)) + + require.Equal(t, []string{"at://A", "at://B"}, rec.startURIs(), + "each object must be tagged with the livestream active when it started") + keys := rec.startKeys() + require.Len(t, keys, 2) + require.NotEqual(t, keys[0], keys[1], "rolled-over objects must have distinct keys") +} + // 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 -- 2.51.2