From 61188f290285f94ec2f2f940c16bc7f291fc1450 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jakub=20Per=C5=BCy=C5=82o?= Date: Fri, 10 Jul 2026 12:28:40 +0200 Subject: [PATCH 1/2] feat(moq-ffi): expose group fetch API Expose single-group fetches by track name and sequence across UniFFI and the language wrappers. Add dynamic group requests for cache misses and document moq-lite 05+ support. Co-authored-by: GPT-5 Codex --- doc/concept/layer/moq-lite.md | 2 +- doc/lib/go/moq.md | 38 +++++ doc/lib/kt/moq.md | 29 ++++ doc/lib/py/moq-rs.md | 27 ++++ doc/lib/swift/moq.md | 29 ++++ go/wrapper/moq/errors.go | 2 + go/wrapper/moq/moq_test.go | 79 ++++++++++ go/wrapper/moq/publish.go | 118 +++++++++++++++ go/wrapper/moq/subscribe.go | 10 ++ go/wrapper/moq/types.go | 1 + .../kotlin/dev/moq/Aliases.kt | 3 + .../jvmAndAndroidMain/kotlin/dev/moq/Flows.kt | 12 ++ .../kotlin/dev/moq/SmokeTest.kt | 21 +++ py/moq-rs/moq/__init__.py | 6 + py/moq-rs/moq/publish.py | 56 +++++++ py/moq-rs/moq/subscribe.py | 25 +++- py/moq-rs/moq/types.py | 4 + py/moq-rs/tests/test_local.py | 28 ++++ rs/moq-ffi/src/consumer.rs | 41 ++++++ rs/moq-ffi/src/error.rs | 8 + rs/moq-ffi/src/producer.rs | 121 +++++++++++++++ rs/moq-ffi/src/test.rs | 139 ++++++++++++++++++ swift/Sources/Moq/Aliases.swift | 2 + swift/Sources/Moq/Broadcast.swift | 10 ++ swift/Sources/Moq/Dynamic.swift | 63 ++++++++ swift/Sources/Moq/Track.swift | 5 + swift/Tests/MoqTests/SmokeTests.swift | 20 +++ 27 files changed, 897 insertions(+), 2 deletions(-) diff --git a/doc/concept/layer/moq-lite.md b/doc/concept/layer/moq-lite.md index 6da72a836c..22826c4b54 100644 --- a/doc/concept/layer/moq-lite.md +++ b/doc/concept/layer/moq-lite.md @@ -163,7 +163,7 @@ But if a publisher needs a feature, then the subscriber needs it too, so you can - **No Request IDs**: A bidirectional stream for each request to avoid HoLB. (NOTE: likely to be upstreamed into moq-transport) - **No Push**: A subscriber must explicitly subscribe to each track. -- **No FETCH**: Use HTTP for VOD instead of reinventing the wheel. +- **Single-group FETCH only (lite-05+)**: Fetch one complete group by sequence. Ranges and joining fetches are not supported. - **No Joining Fetch**: Subscriptions start at the latest group, not the latest frame. - **No sub-groups**: SVC layers should be separate tracks. - **No gaps**: Makes life much easier for the relay and every application. diff --git a/doc/lib/go/moq.md b/doc/lib/go/moq.md index 365f29cd50..2e2369ee97 100644 --- a/doc/lib/go/moq.md +++ b/doc/lib/go/moq.md @@ -89,6 +89,44 @@ broadcast.Finish() If a producer is collected without `Finish()`, the underlying library logs a warning (`broadcast::Producer dropped without close()`) to help you spot the leak. +## Fetching raw groups + +Fetch retrieves one group by track name and group sequence without keeping a live subscription: + +```go +group, err := consumer.FetchGroup("events", 42, &moq.FetchGroupOptions{Priority: 10}) +if err != nil { + log.Fatal(err) +} +for frame, err := range group.Frames(ctx) { + if err != nil { + log.Fatal(err) + } + fmt.Printf("%s\n", frame) +} +``` + +A retained group resolves immediately. To serve a group that is not retained, keep a dynamic handler alive on its producer: + +```go +dynamic, err := track.Dynamic() +if err != nil { + log.Fatal(err) +} +request, err := dynamic.RequestedGroup(ctx) +if err != nil { + log.Fatal(err) +} +producer, err := request.Accept() +if err != nil { + log.Fatal(err) +} +_ = producer.WriteFrame(loadArchivedFrame(request.Sequence())) +_ = producer.Finish() +``` + +Call `request.Abort(code)` when the requested group cannot be produced. Fetch is currently a single-group operation and is supported by the moq-lite 05+ FETCH wire path. + ## Local development The in-tree `go/wrapper/` directory is the source skeleton; CI publishes it to the [moq-dev/moq-go](https://github.com/moq-dev/moq-go) mirror. To exercise it locally: diff --git a/doc/lib/kt/moq.md b/doc/lib/kt/moq.md index 736fef2857..a52b8c8321 100644 --- a/doc/lib/kt/moq.md +++ b/doc/lib/kt/moq.md @@ -89,6 +89,35 @@ Moq.connect("https://relay.example.com").use { moq -> } ``` +### Fetching raw groups + +Fetch retrieves one group by track name and group sequence without keeping a live subscription: + +```kotlin +val group = consumer.fetchGroup( + "events", + 42uL, + FetchGroupOptions(priority = 10u), +) +group.frames().collect { frame -> + println(frame.decodeToString()) +} +``` + +A retained group resolves immediately. To serve a group that is not retained, keep a dynamic handler alive on its producer: + +```kotlin +val dynamic = track.dynamic() + +dynamic.requestedGroups().collect { request -> + val group = request.accept() + group.writeFrame(loadArchivedFrame(request.sequence())) + group.finish() +} +``` + +Call `request.abort(code)` when the requested group cannot be produced. Fetch is currently a single-group operation and is supported by the moq-lite 05+ FETCH wire path. + ### On-demand raw tracks Use a dynamic broadcast when subscribers should be able to request raw tracks that are not published yet: diff --git a/doc/lib/py/moq-rs.md b/doc/lib/py/moq-rs.md index b926d121d4..42ff7bc5b0 100644 --- a/doc/lib/py/moq-rs.md +++ b/doc/lib/py/moq-rs.md @@ -122,6 +122,33 @@ async for group in track: `write_frame` on a track creates a one-frame group by default. Use `append_group()` for multi-frame groups (e.g., a video GOP). +### Fetching raw groups + +Fetch retrieves one group by track name and group sequence without keeping a live subscription: + +```python +group = await broadcast_consumer.fetch_group( + "events", + sequence=42, + options=moq.FetchGroupOptions(priority=10), +) +async for frame in group: + print(frame) +``` + +A retained group resolves immediately. To serve a group that is not retained, keep a dynamic handler alive on its producer: + +```python +dynamic = track.dynamic() + +async for request in dynamic: + group = request.accept() + group.write_frame(load_archived_frame(request.sequence)) + group.finish() +``` + +Call `request.abort(code)` when the requested group cannot be produced. Fetch is currently a single-group operation and is supported by the moq-lite 05+ FETCH wire path. + ### On-demand raw tracks Use a dynamic broadcast when subscribers should be able to request raw tracks that are not published yet: diff --git a/doc/lib/swift/moq.md b/doc/lib/swift/moq.md index 6f1f510eb7..ebde783d76 100644 --- a/doc/lib/swift/moq.md +++ b/doc/lib/swift/moq.md @@ -95,6 +95,35 @@ try audio.finish() try broadcast.finish() ``` +### Fetching raw groups + +Fetch retrieves one group by track name and group sequence without keeping a live subscription: + +```swift +let group = try await consumer.fetchGroup( + name: "events", + sequence: 42, + options: FetchGroupOptions(priority: 10) +) +for try await frame in group { + print(frame) +} +``` + +A retained group resolves immediately. To serve a group that is not retained, keep a dynamic handler alive on its producer: + +```swift +let dynamic = try track.dynamic() + +for try await request in dynamic { + let group = try request.accept() + try group.writeFrame(loadArchivedFrame(request.sequence)) + try group.finish() +} +``` + +Call `request.abort(errorCode:)` when the requested group cannot be produced. Fetch is currently a single-group operation and is supported by the moq-lite 05+ FETCH wire path. + ### On-demand raw tracks Use a dynamic broadcast when subscribers should be able to request raw tracks that are not published yet: diff --git a/go/wrapper/moq/errors.go b/go/wrapper/moq/errors.go index 99b77f6fad..b6cbf57af3 100644 --- a/go/wrapper/moq/errors.go +++ b/go/wrapper/moq/errors.go @@ -42,6 +42,8 @@ var ( ErrInvalidErrorCode = ffi.ErrMoqErrorInvalidErrorCode ErrUnauthorized = ffi.ErrMoqErrorUnauthorized ErrForbidden = ffi.ErrMoqErrorForbidden + ErrNotFound = ffi.ErrMoqErrorNotFound + ErrUnsupported = ffi.ErrMoqErrorUnsupported ErrLog = ffi.ErrMoqErrorLog ) diff --git a/go/wrapper/moq/moq_test.go b/go/wrapper/moq/moq_test.go index 367d45161b..da19d06f43 100644 --- a/go/wrapper/moq/moq_test.go +++ b/go/wrapper/moq/moq_test.go @@ -51,6 +51,85 @@ func TestPublishMediaLifecycle(t *testing.T) { } } +func TestFetchGroupAndServeDynamicMiss(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), testTimeout) + defer cancel() + + broadcast, err := moq.NewBroadcastProducer() + if err != nil { + t.Fatal(err) + } + track, err := broadcast.PublishTrack("events", nil) + if err != nil { + t.Fatal(err) + } + consumer, err := broadcast.Consume() + if err != nil { + t.Fatal(err) + } + + cached, err := track.AppendGroup() + if err != nil { + t.Fatal(err) + } + if err := cached.WriteFrame([]byte("cached")); err != nil { + t.Fatal(err) + } + if err := cached.Finish(); err != nil { + t.Fatal(err) + } + + fetched, err := consumer.FetchGroup("events", 0, &moq.FetchGroupOptions{Priority: 3}) + if err != nil { + t.Fatal(err) + } + frame, err := fetched.ReadFrame(ctx) + if err != nil || string(frame) != "cached" { + t.Fatalf("cached fetch: frame=%q err=%v", frame, err) + } + + dynamic, err := track.Dynamic() + if err != nil { + t.Fatal(err) + } + type fetchResult struct { + group *moq.GroupConsumer + err error + } + result := make(chan fetchResult, 1) + go func() { + group, err := consumer.FetchGroup("events", 7, &moq.FetchGroupOptions{Priority: 11}) + result <- fetchResult{group: group, err: err} + }() + + request, err := dynamic.RequestedGroup(ctx) + if err != nil { + t.Fatal(err) + } + if request.Sequence() != 7 || request.Priority() != 11 { + t.Fatalf("unexpected request: sequence=%d priority=%d", request.Sequence(), request.Priority()) + } + produced, err := request.Accept() + if err != nil { + t.Fatal(err) + } + if err := produced.WriteFrame([]byte("archive")); err != nil { + t.Fatal(err) + } + if err := produced.Finish(); err != nil { + t.Fatal(err) + } + + res := <-result + if res.err != nil { + t.Fatal(res.err) + } + frame, err = res.group.ReadFrame(ctx) + if err != nil || string(frame) != "archive" { + t.Fatalf("dynamic fetch: frame=%q err=%v", frame, err) + } +} + func TestUnknownFormat(t *testing.T) { broadcast, err := moq.NewBroadcastProducer() if err != nil { diff --git a/go/wrapper/moq/publish.go b/go/wrapper/moq/publish.go index 89b02ac3c8..e08e204e39 100644 --- a/go/wrapper/moq/publish.go +++ b/go/wrapper/moq/publish.go @@ -71,6 +71,15 @@ func (b *BroadcastProducer) Consume() (*BroadcastConsumer, error) { return &BroadcastConsumer{inner: inner}, nil } +// Dynamic accepts requests for tracks that are not published yet. +func (b *BroadcastProducer) Dynamic() (*BroadcastDynamic, error) { + inner, err := b.inner.Dynamic() + if err != nil { + return nil, err + } + return &BroadcastDynamic{inner: inner}, nil +} + // SetCatalogSection sets (or replaces) an untyped application catalog section by // name. json is any JSON document as a string; it rides alongside video/audio and // reaches subscribers via Catalog.Sections. name must not be a reserved media @@ -161,6 +170,15 @@ func (t *TrackProducer) Unused(ctx context.Context) error { return runErr(ctx, nil, t.inner.Unused) } +// Dynamic serves fetches for groups that are not currently cached. +func (t *TrackProducer) Dynamic() (*TrackDynamic, error) { + inner, err := t.inner.Dynamic() + if err != nil { + return nil, err + } + return &TrackDynamic{inner: inner}, nil +} + // AppendGroup starts a new group; write frames into it, then Finish. func (t *TrackProducer) AppendGroup() (*GroupProducer, error) { inner, err := t.inner.AppendGroup() @@ -219,6 +237,106 @@ func (g *GroupProducer) Finish() error { return g.inner.Finish() } +// BroadcastDynamic yields tracks requested by subscribers. +type BroadcastDynamic struct { + inner *ffi.MoqBroadcastDynamic +} + +// RequestedTrack waits for the next requested track. +func (d *BroadcastDynamic) RequestedTrack(ctx context.Context) (*TrackRequest, error) { + inner, err := runCancellable(ctx, d.inner.Cancel, d.inner.RequestedTrack) + if err != nil { + return nil, err + } + return &TrackRequest{inner: inner}, nil +} + +// Cancel stops current and future requested-track waits. +func (d *BroadcastDynamic) Cancel() { + d.inner.Cancel() +} + +// TrackRequest is a subscriber-requested track that has not been accepted yet. +type TrackRequest struct { + inner *ffi.MoqTrackRequest +} + +// Name is the requested track name. +func (r *TrackRequest) Name() (string, error) { + return r.inner.Name() +} + +// Dynamic creates a fetch handler before accepting this requested track. +func (r *TrackRequest) Dynamic() (*TrackDynamic, error) { + inner, err := r.inner.Dynamic() + if err != nil { + return nil, err + } + return &TrackDynamic{inner: inner}, nil +} + +// Accept accepts the request as a raw track. +func (r *TrackRequest) Accept(info *TrackInfo) (*TrackProducer, error) { + inner, err := r.inner.Accept(info) + if err != nil { + return nil, err + } + return &TrackProducer{inner: inner}, nil +} + +// Abort rejects the requested track with an application error code. +func (r *TrackRequest) Abort(errorCode int32) error { + return r.inner.Abort(errorCode) +} + +// TrackDynamic yields uncached groups requested by fetch consumers. +type TrackDynamic struct { + inner *ffi.MoqTrackDynamic +} + +// RequestedGroup waits for the next uncached group request. +func (d *TrackDynamic) RequestedGroup(ctx context.Context) (*GroupRequest, error) { + inner, err := runCancellable(ctx, d.inner.Cancel, d.inner.RequestedGroup) + if err != nil { + return nil, err + } + return &GroupRequest{inner: inner}, nil +} + +// Cancel stops current and future requested-group waits. +func (d *TrackDynamic) Cancel() { + d.inner.Cancel() +} + +// GroupRequest requests one uncached group from a track producer. +type GroupRequest struct { + inner *ffi.MoqGroupRequest +} + +// Sequence is the requested group sequence within the track. +func (r *GroupRequest) Sequence() uint64 { + return r.inner.Sequence() +} + +// Priority is the consumer's delivery priority for this fetch. +func (r *GroupRequest) Priority() uint8 { + return r.inner.Priority() +} + +// Accept accepts the request and returns a producer for the group. +func (r *GroupRequest) Accept() (*GroupProducer, error) { + inner, err := r.inner.Accept() + if err != nil { + return nil, err + } + return &GroupProducer{inner: inner}, nil +} + +// Abort rejects the fetch with an application error code. +func (r *GroupRequest) Abort(errorCode int32) error { + return r.inner.Abort(errorCode) +} + // AudioProducer pushes raw PCM and lets libopus encode it on the way out. type AudioProducer struct { inner *ffi.MoqAudioProducer diff --git a/go/wrapper/moq/subscribe.go b/go/wrapper/moq/subscribe.go index 6f93c67338..cc17fb9261 100644 --- a/go/wrapper/moq/subscribe.go +++ b/go/wrapper/moq/subscribe.go @@ -31,6 +31,16 @@ func (b *BroadcastConsumer) SubscribeTrack(name string, subscription *Subscripti return &TrackConsumer{inner: inner}, nil } +// FetchGroup fetches one complete group by track name and group sequence +// without holding a live subscription. +func (b *BroadcastConsumer) FetchGroup(name string, sequence uint64, options *FetchGroupOptions) (*GroupConsumer, error) { + inner, err := b.inner.FetchGroup(name, sequence, options) + if err != nil { + return nil, err + } + return &GroupConsumer{inner: inner}, nil +} + // SubscribeMedia subscribes to a media track, decoded with the given container. // maxLatencyMs bounds buffering before a stalled group is skipped. subscription // tunes delivery (priority, ordering, group range); pass nil for defaults. diff --git a/go/wrapper/moq/types.go b/go/wrapper/moq/types.go index b6b8bb9b52..082557dc01 100644 --- a/go/wrapper/moq/types.go +++ b/go/wrapper/moq/types.go @@ -16,6 +16,7 @@ type ( Catalog = ffi.MoqCatalog Dimensions = ffi.MoqDimensions Frame = ffi.MoqFrame + FetchGroupOptions = ffi.MoqFetchGroupOptions Subscription = ffi.MoqSubscription TrackInfo = ffi.MoqTrackInfo Video = ffi.MoqVideo diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt index caa913c443..c325acdfe4 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt @@ -25,7 +25,9 @@ typealias BroadcastProducer = uniffi.moq.MoqBroadcastProducer typealias BroadcastConsumer = uniffi.moq.MoqBroadcastConsumer typealias TrackProducer = uniffi.moq.MoqTrackProducer typealias TrackRequest = uniffi.moq.MoqTrackRequest +typealias TrackDynamic = uniffi.moq.MoqTrackDynamic typealias TrackConsumer = uniffi.moq.MoqTrackConsumer +typealias GroupRequest = uniffi.moq.MoqGroupRequest typealias GroupProducer = uniffi.moq.MoqGroupProducer typealias GroupConsumer = uniffi.moq.MoqGroupConsumer @@ -44,6 +46,7 @@ typealias Video = uniffi.moq.MoqVideo typealias Audio = uniffi.moq.MoqAudio typealias Dimensions = uniffi.moq.MoqDimensions typealias Subscription = uniffi.moq.MoqSubscription +typealias FetchGroupOptions = uniffi.moq.MoqFetchGroupOptions typealias TrackInfo = uniffi.moq.MoqTrackInfo typealias AudioFrame = uniffi.moq.MoqAudioFrame typealias AudioCodec = uniffi.moq.MoqAudioCodec diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Flows.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Flows.kt index 1549b1f88f..12a37b9c56 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Flows.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Flows.kt @@ -16,9 +16,11 @@ import uniffi.moq.MoqCatalogConsumer import uniffi.moq.MoqException import uniffi.moq.MoqFrame import uniffi.moq.MoqGroupConsumer +import uniffi.moq.MoqGroupRequest import uniffi.moq.MoqMediaConsumer import uniffi.moq.MoqOriginConsumer import uniffi.moq.MoqTrackConsumer +import uniffi.moq.MoqTrackDynamic import uniffi.moq.MoqTrackRequest /** @@ -104,6 +106,16 @@ fun MoqBroadcastDynamic.requestedTracks(): Flow = flow { if (cause is CancellationException) cancel() } +/** Stream of uncached group requests for one track. */ +fun MoqTrackDynamic.requestedGroups(): Flow = flow { + while (true) { + currentCoroutineContext().ensureActive() + emit(requestedGroup()) + } +}.onCompletion { cause -> + if (cause is CancellationException) cancel() +} + /** Stream of raw frame payloads within a group. */ fun MoqGroupConsumer.frames(): Flow = flow { while (true) { diff --git a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt index d8157003dd..b36e503e5c 100644 --- a/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt +++ b/kt/moq/src/jvmAndAndroidTest/kotlin/dev/moq/SmokeTest.kt @@ -3,7 +3,9 @@ package dev.moq import kotlinx.coroutines.test.runTest import uniffi.moq.MoqException import kotlin.test.Test +import kotlin.test.assertEquals import kotlin.test.assertFailsWith +import kotlin.test.assertNull import kotlin.test.assertTrue class SmokeTest { @@ -34,4 +36,23 @@ class SmokeTest { origin.consume().use { /* lifecycle smoke */ } } } + + @Test + fun `broadcast consumer fetches cached group`() = runTest { + BroadcastProducer().use { broadcast -> + val track = broadcast.publishTrack("events", null) + val group = track.appendGroup() + group.writeFrame("cached".encodeToByteArray()) + group.finish() + + val fetched = broadcast.consume().fetchGroup( + "events", + 0uL, + FetchGroupOptions(priority = 3u), + ) + assertEquals(0uL, fetched.sequence()) + assertEquals("cached", fetched.readFrame()?.decodeToString()) + assertNull(fetched.readFrame()) + } + } } diff --git a/py/moq-rs/moq/__init__.py b/py/moq-rs/moq/__init__.py index f677d33dde..334789e24f 100644 --- a/py/moq-rs/moq/__init__.py +++ b/py/moq-rs/moq/__init__.py @@ -13,8 +13,10 @@ BroadcastDynamic, BroadcastProducer, GroupProducer, + GroupRequest, MediaProducer, MediaStreamProducer, + TrackDynamic, TrackProducer, TrackRequest, ) @@ -40,6 +42,7 @@ ConnectionStats, Container, Dimensions, + FetchGroupOptions, Frame, Subscription, TrackInfo, @@ -71,7 +74,9 @@ "Dimensions", "Error", "Frame", + "FetchGroupOptions", "GroupConsumer", + "GroupRequest", "GroupProducer", "MediaConsumer", "MediaProducer", @@ -83,6 +88,7 @@ "Session", "Subscription", "TrackConsumer", + "TrackDynamic", "TrackInfo", "TrackProducer", "TrackRequest", diff --git a/py/moq-rs/moq/publish.py b/py/moq-rs/moq/publish.py index 16efa17d1f..5b8849454e 100644 --- a/py/moq-rs/moq/publish.py +++ b/py/moq-rs/moq/publish.py @@ -9,9 +9,11 @@ MoqBroadcastDynamic, MoqBroadcastProducer, MoqGroupProducer, + MoqGroupRequest, MoqInit, MoqMediaProducer, MoqMediaStreamProducer, + MoqTrackDynamic, MoqTrackProducer, MoqTrackRequest, ) @@ -118,6 +120,10 @@ async def unused(self) -> None: """Wait until this track has no active subscribers.""" await self._inner.unused() + def dynamic(self) -> TrackDynamic: + """Serve fetches for groups that are not currently cached.""" + return TrackDynamic(self._inner.dynamic()) + def append_group(self) -> GroupProducer: """Start a new group; write frames into it, then finish().""" return GroupProducer(self._inner.append_group()) @@ -165,11 +171,61 @@ def accept(self, info: TrackInfo | None = None) -> TrackProducer: """ return TrackProducer(self._inner.accept(info)) + def dynamic(self) -> TrackDynamic: + """Create a fetch handler before accepting this requested track.""" + return TrackDynamic(self._inner.dynamic()) + def abort(self, error_code: int) -> None: """Reject the request with an application error code.""" self._inner.abort(error_code) +class GroupRequest: + """A request to produce one uncached group for a fetch consumer.""" + + def __init__(self, inner: MoqGroupRequest) -> None: + self._inner = inner + + @property + def sequence(self) -> int: + """The requested group sequence within the track.""" + return self._inner.sequence() + + @property + def priority(self) -> int: + """The consumer's delivery priority for this fetch.""" + return self._inner.priority() + + def accept(self) -> GroupProducer: + """Accept the request and return a producer for the group.""" + return GroupProducer(self._inner.accept()) + + def abort(self, error_code: int) -> None: + """Reject the fetch with an application error code.""" + self._inner.abort(error_code) + + +class TrackDynamic: + """Async source of uncached group requests for one track.""" + + def __init__(self, inner: MoqTrackDynamic) -> None: + self._inner = inner + + def __aiter__(self): + return self + + async def __anext__(self) -> GroupRequest: + return await self.requested_group() + + async def requested_group(self) -> GroupRequest: + """Wait for the next uncached group request.""" + return GroupRequest(await self._inner.requested_group()) + + def cancel(self) -> None: + """Cancel current and future group request waits.""" + self._inner.cancel() + + class AudioProducer: """Publish raw PCM and let libopus encode it on the way out. diff --git a/py/moq-rs/moq/subscribe.py b/py/moq-rs/moq/subscribe.py index acad8259da..1265d19d8b 100644 --- a/py/moq-rs/moq/subscribe.py +++ b/py/moq-rs/moq/subscribe.py @@ -11,7 +11,17 @@ MoqTrackConsumer, ) -from .types import Audio, AudioDecoderOutput, AudioFrame, Catalog, Container, Frame, Subscription, Video +from .types import ( + Audio, + AudioDecoderOutput, + AudioFrame, + Catalog, + Container, + FetchGroupOptions, + Frame, + Subscription, + Video, +) class MediaConsumer: @@ -200,6 +210,19 @@ async def subscribe_track(self, name: str, subscription: Subscription | None = N """ return TrackConsumer(await self._inner.subscribe_track(name, subscription)) + async def fetch_group( + self, + name: str, + sequence: int, + options: FetchGroupOptions | None = None, + ) -> GroupConsumer: + """Fetch one complete group by track name and group sequence. + + This does not hold a live subscription. The returned group may still be + receiving frames, so iterate it until completion. + """ + return GroupConsumer(await self._inner.fetch_group(name, sequence, options)) + async def subscribe_media( self, name: str, diff --git a/py/moq-rs/moq/types.py b/py/moq-rs/moq/types.py index 89ed8eb822..2bf1a2eeff 100644 --- a/py/moq-rs/moq/types.py +++ b/py/moq-rs/moq/types.py @@ -33,6 +33,9 @@ from moq_ffi import ( MoqDimensions as Dimensions, ) +from moq_ffi import ( + MoqFetchGroupOptions as FetchGroupOptions, +) from moq_ffi import ( MoqFrame as Frame, ) @@ -62,6 +65,7 @@ "Container", "Dimensions", "Frame", + "FetchGroupOptions", "Subscription", "TrackInfo", "Video", diff --git a/py/moq-rs/tests/test_local.py b/py/moq-rs/tests/test_local.py index 23be888bc8..92c4560d2b 100644 --- a/py/moq-rs/tests/test_local.py +++ b/py/moq-rs/tests/test_local.py @@ -254,6 +254,34 @@ async def test_publish_track_info_and_subscription(): track.finish() +async def test_fetch_group_and_serve_dynamic_miss(): + """Fetch a cached group, then serve an uncached sequence through TrackDynamic.""" + broadcast = moq.BroadcastProducer() + track = broadcast.publish_track("events") + consumer = broadcast.consume() + + cached = track.append_group() + cached.write_frame(b"cached") + cached.finish() + + fetched = await consumer.fetch_group("events", 0, moq.FetchGroupOptions(priority=3)) + assert fetched.sequence == 0 + assert [frame async for frame in fetched] == [b"cached"] + + dynamic = track.dynamic() + pending = asyncio.create_task(consumer.fetch_group("events", 7, moq.FetchGroupOptions(priority=11))) + request = await asyncio.wait_for(dynamic.requested_group(), timeout=5.0) + assert request.sequence == 7 + assert request.priority == 11 + + produced = request.accept() + produced.write_frame(b"archive") + produced.finish() + + fetched = await asyncio.wait_for(pending, timeout=5.0) + assert [frame async for frame in fetched] == [b"archive"] + + async def test_dynamic_track_request(): broadcast = moq.BroadcastProducer() dynamic = broadcast.dynamic() diff --git a/rs/moq-ffi/src/consumer.rs b/rs/moq-ffi/src/consumer.rs index e1621167a7..dd28a3562e 100644 --- a/rs/moq-ffi/src/consumer.rs +++ b/rs/moq-ffi/src/consumer.rs @@ -30,6 +30,22 @@ pub struct MoqSubscription { pub group_end: Option, } +/// Options for fetching one past group by sequence. +#[derive(Clone, uniffi::Record)] +pub struct MoqFetchGroupOptions { + /// Delivery priority for the fetch stream; higher values preempt lower ones. + #[uniffi(default = 0)] + pub priority: u8, +} + +impl From for moq_net::group::Fetch { + fn from(options: MoqFetchGroupOptions) -> Self { + Self { + priority: options.priority, + } + } +} + impl From for moq_net::Subscription { fn from(s: MoqSubscription) -> Self { moq_net::Subscription::default() @@ -144,6 +160,23 @@ impl MoqBroadcastConsumer { Ok(Arc::new(MoqTrackConsumer::new(track))) } + /// Fetch one complete group by track name and group sequence. + /// + /// This does not create a live subscription. A retained group resolves immediately; + /// otherwise the request waits for a dynamic producer to serve it. The returned + /// group may still be in progress, so read frames until `read_frame()` returns `None`. + pub async fn fetch_group( + &self, + name: String, + sequence: u64, + options: Option, + ) -> Result, MoqError> { + let options = options.map(moq_net::group::Fetch::from); + let track = self.inner.track(&name).map_err(map_fetch_error)?; + let group = track.fetch_group(sequence, options).await.map_err(map_fetch_error)?; + Ok(Arc::new(MoqGroupConsumer::new(group))) + } + /// Subscribe to a track by name, delivering frames in decode order. /// /// `container` is the track container from the catalog. @@ -172,6 +205,14 @@ impl MoqBroadcastConsumer { } } +fn map_fetch_error(err: moq_net::Error) -> MoqError { + match err { + moq_net::Error::NotFound => MoqError::NotFound, + moq_net::Error::Unsupported | moq_net::Error::Version => MoqError::Unsupported, + err => err.into(), + } +} + // ---- Track Consumer ---- struct TrackInner { diff --git a/rs/moq-ffi/src/error.rs b/rs/moq-ffi/src/error.rs index 1b5af44660..00ec97ae51 100644 --- a/rs/moq-ffi/src/error.rs +++ b/rs/moq-ffi/src/error.rs @@ -60,6 +60,14 @@ pub enum MoqError { #[error("forbidden")] Forbidden, + /// The requested track or group is not available. + #[error("not found")] + NotFound, + + /// The negotiated protocol does not support the requested operation. + #[error("unsupported")] + Unsupported, + #[error("log: {0}")] Log(String), } diff --git a/rs/moq-ffi/src/producer.rs b/rs/moq-ffi/src/producer.rs index 773e2d2c51..05b308e2f2 100644 --- a/rs/moq-ffi/src/producer.rs +++ b/rs/moq-ffi/src/producer.rs @@ -130,10 +130,28 @@ pub struct MoqBroadcastDynamic { task: Task, } +/// Serves on-demand fetches of uncached groups for one track. +#[derive(uniffi::Object)] +pub struct MoqTrackDynamic { + task: Task, +} + +impl MoqTrackDynamic { + fn new(inner: moq_net::track::Dynamic) -> Self { + Self { + task: Task::new(TrackDynamicProducer { inner }), + } + } +} + struct DynamicProducer { inner: moq_net::broadcast::Dynamic, } +struct TrackDynamicProducer { + inner: moq_net::track::Dynamic, +} + impl DynamicProducer { async fn requested_track(&mut self) -> Result, MoqError> { // Hand back the un-accepted request, mirroring `moq_net::broadcast::Dynamic`: the caller @@ -144,6 +162,13 @@ impl DynamicProducer { } } +impl TrackDynamicProducer { + async fn requested_group(&mut self) -> Result, MoqError> { + let request = self.inner.requested_group().await?; + Ok(Arc::new(MoqGroupRequest::new(request))) + } +} + impl MoqBroadcastProducer { pub(crate) fn consume_inner(&self) -> Result { let guard = self.state.lock().unwrap(); @@ -407,6 +432,79 @@ impl MoqBroadcastDynamic { } } +// ---- Dynamic Track Producer ---- + +#[uniffi::export] +impl MoqTrackDynamic { + /// Wait for the next fetch of an uncached group. + /// + /// Accept the returned request to produce the group, or abort it with an + /// application error. Cached groups are served without reaching this method. + pub async fn requested_group(&self) -> Result, MoqError> { + self.task + .run(|mut state| async move { state.requested_group().await }) + .await + } + + /// Cancel all current and future `requested_group()` calls. + pub fn cancel(&self) { + self.task.cancel(); + } +} + +/// An uncached group requested by a fetch consumer. +#[derive(uniffi::Object)] +pub struct MoqGroupRequest { + sequence: u64, + priority: u8, + inner: std::sync::Mutex>, +} + +impl MoqGroupRequest { + fn new(request: moq_net::track::GroupRequest) -> Self { + Self { + sequence: request.sequence(), + priority: request.priority(), + inner: std::sync::Mutex::new(Some(request)), + } + } + + fn take(&self) -> Result { + self.inner.lock().unwrap().take().ok_or(MoqError::Closed) + } +} + +#[uniffi::export] +impl MoqGroupRequest { + /// The requested group sequence within the track. + pub fn sequence(&self) -> u64 { + self.sequence + } + + /// The consumer's delivery priority for this fetch. + pub fn priority(&self) -> u8 { + self.priority + } + + /// Accept the request and return a producer for filling the fetched group. + pub fn accept(&self) -> Result, MoqError> { + let _guard = crate::ffi::RUNTIME.enter(); + let group = self.take()?.accept(None)?; + Ok(Arc::new(MoqGroupProducer { + sequence: group.sequence, + inner: std::sync::Mutex::new(Some(group)), + })) + } + + /// Reject the fetch with an application error code. + pub fn abort(&self, error_code: i32) -> Result<(), MoqError> { + let _guard = crate::ffi::RUNTIME.enter(); + let error_code = u16::try_from(error_code).map_err(|_| MoqError::InvalidErrorCode(error_code))?; + self.take()?.reject(moq_net::Error::App(error_code)); + Ok(()) + } +} + // ---- Track Request ---- /// A track requested by a subscriber that hasn't been accepted yet. @@ -442,6 +540,18 @@ impl MoqTrackRequest { Ok(request.name().to_string()) } + /// Create a handler for uncached group fetches before accepting this track. + /// + /// Obtain and retain this handle before `accept()` when the track itself was + /// requested by a fetch. This keeps the pending group request serviceable across + /// the transition from request to producer. + pub fn dynamic(&self) -> Result, MoqError> { + let _guard = crate::ffi::RUNTIME.enter(); + let guard = self.inner.lock().unwrap(); + let request = guard.as_ref().ok_or(MoqError::Closed)?; + Ok(Arc::new(MoqTrackDynamic::new(request.dynamic()))) + } + /// Accept the request as a raw track, fixing its [`MoqTrackInfo`] (timescale, etc.). /// /// For media use [`MoqBroadcastProducer::publish_media_on_track`] instead, which lets @@ -482,6 +592,17 @@ impl MoqTrackProducer { Ok(track.name().to_string()) } + /// Create a handler for uncached group fetches on this track. + /// + /// Hold the returned object for as long as cache misses should wait to be + /// served. Without a live dynamic handler, a missing group fails with `NotFound`. + pub fn dynamic(&self) -> Result, MoqError> { + let _guard = crate::ffi::RUNTIME.enter(); + let guard = self.inner.lock().unwrap(); + let track = guard.as_ref().ok_or(MoqError::Closed)?; + Ok(Arc::new(MoqTrackDynamic::new(track.dynamic()))) + } + /// Wait until this track has at least one active consumer. pub async fn used(&self) -> Result<(), MoqError> { let track = self.inner.lock().unwrap().as_ref().ok_or(MoqError::Closed)?.clone(); diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index bd3c600d19..23f744460e 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -2,6 +2,7 @@ use super::origin::*; use super::producer::*; use super::server::MoqServer; use super::session::MoqClient; +use crate::consumer::MoqFetchGroupOptions; use crate::error::MoqError; use crate::media::MoqInit; @@ -149,6 +150,144 @@ async fn dynamic_track_request_can_abort() { assert!(result.is_err(), "subscribe to a rejected track should fail"); } +#[tokio::test] +async fn fetches_cached_group_without_subscribing() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let track = broadcast.publish_track("events".into(), None).unwrap(); + let group = track.append_group().unwrap(); + group.write_frame(b"first".to_vec()).unwrap(); + group.write_frame(b"second".to_vec()).unwrap(); + group.finish().unwrap(); + + let consumer = broadcast.consume().unwrap(); + let fetched = consumer + .fetch_group("events".into(), 0, Some(MoqFetchGroupOptions { priority: 7 })) + .await + .unwrap(); + + assert_eq!(fetched.sequence(), 0); + assert_eq!( + fetched.read_frame().await.unwrap().as_deref(), + Some(b"first".as_slice()) + ); + assert_eq!( + fetched.read_frame().await.unwrap().as_deref(), + Some(b"second".as_slice()) + ); + assert_eq!(fetched.read_frame().await.unwrap(), None); +} + +#[tokio::test] +async fn dynamic_track_serves_fetch_miss_and_priority() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let track = broadcast.publish_track("events".into(), None).unwrap(); + let dynamic = track.dynamic().unwrap(); + let consumer = broadcast.consume().unwrap(); + + let fetch = tokio::spawn(async move { + consumer + .fetch_group("events".into(), 5, Some(MoqFetchGroupOptions { priority: 11 })) + .await + }); + + let request = tokio::time::timeout(TIMEOUT, dynamic.requested_group()) + .await + .expect("timed out waiting for group request") + .unwrap(); + assert_eq!(request.sequence(), 5); + assert_eq!(request.priority(), 11); + + let group = request.accept().unwrap(); + group.write_frame(b"fetched".to_vec()).unwrap(); + group.finish().unwrap(); + + let fetched = tokio::time::timeout(TIMEOUT, fetch) + .await + .expect("timed out waiting for fetch") + .expect("fetch task panicked") + .unwrap(); + assert_eq!(fetched.sequence(), 5); + assert_eq!( + fetched.read_frame().await.unwrap().as_deref(), + Some(b"fetched".as_slice()) + ); +} + +#[tokio::test] +async fn dynamic_track_rejects_fetch_miss() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let track = broadcast.publish_track("events".into(), None).unwrap(); + let dynamic = track.dynamic().unwrap(); + let consumer = broadcast.consume().unwrap(); + + let fetch = tokio::spawn(async move { consumer.fetch_group("events".into(), 5, None).await }); + let request = tokio::time::timeout(TIMEOUT, dynamic.requested_group()) + .await + .expect("timed out waiting for group request") + .unwrap(); + request.abort(404).unwrap(); + + let result = tokio::time::timeout(TIMEOUT, fetch) + .await + .expect("timed out waiting for rejected fetch") + .expect("fetch task panicked"); + assert!(matches!(result, Err(MoqError::Protocol(moq_net::Error::App(404))))); + assert!(matches!(request.accept(), Err(MoqError::Closed))); +} + +#[tokio::test] +async fn fetch_miss_without_dynamic_is_not_found() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let _track = broadcast.publish_track("events".into(), None).unwrap(); + let consumer = broadcast.consume().unwrap(); + + let result = consumer.fetch_group("events".into(), 5, None).await; + assert!(matches!(result, Err(MoqError::NotFound))); +} + +#[tokio::test] +async fn fetch_unknown_track_is_not_found() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let consumer = broadcast.consume().unwrap(); + + let result = consumer.fetch_group("missing".into(), 0, None).await; + assert!(matches!(result, Err(MoqError::NotFound))); +} + +#[tokio::test] +async fn requested_track_dynamic_survives_accept() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let broadcast_dynamic = broadcast.dynamic().unwrap(); + let consumer = broadcast.consume().unwrap(); + + let fetch = tokio::spawn(async move { consumer.fetch_group("archive".into(), 9, None).await }); + let request = tokio::time::timeout(TIMEOUT, broadcast_dynamic.requested_track()) + .await + .expect("timed out waiting for track request") + .unwrap(); + let track_dynamic = request.dynamic().unwrap(); + let _track = request.accept(None).unwrap(); + + let group_request = tokio::time::timeout(TIMEOUT, track_dynamic.requested_group()) + .await + .expect("timed out waiting for group request") + .unwrap(); + assert_eq!(group_request.sequence(), 9); + let group = group_request.accept().unwrap(); + group.write_frame(b"archive".to_vec()).unwrap(); + group.finish().unwrap(); + + let fetched = tokio::time::timeout(TIMEOUT, fetch) + .await + .expect("timed out waiting for fetch") + .expect("fetch task panicked") + .unwrap(); + assert_eq!( + fetched.read_frame().await.unwrap().as_deref(), + Some(b"archive".as_slice()) + ); +} + #[tokio::test] async fn dynamic_track_request_can_publish_media() { let broadcast = MoqBroadcastProducer::new().unwrap(); diff --git a/swift/Sources/Moq/Aliases.swift b/swift/Sources/Moq/Aliases.swift index 4d7161f9fd..19adfa1e8e 100644 --- a/swift/Sources/Moq/Aliases.swift +++ b/swift/Sources/Moq/Aliases.swift @@ -19,6 +19,8 @@ public typealias AudioFormat = MoqFFI.MoqAudioFormat public typealias AudioCodec = MoqFFI.MoqAudioCodec public typealias Container = MoqFFI.Container public typealias Subscription = MoqFFI.MoqSubscription +/// Options for fetching one complete group by sequence. +public typealias FetchGroupOptions = MoqFFI.MoqFetchGroupOptions public typealias TrackInfo = MoqFFI.MoqTrackInfo /// A snapshot of connection statistics (RTT, bandwidth estimates, byte/packet diff --git a/swift/Sources/Moq/Broadcast.swift b/swift/Sources/Moq/Broadcast.swift index 99bdb0ce82..8ea939786d 100644 --- a/swift/Sources/Moq/Broadcast.swift +++ b/swift/Sources/Moq/Broadcast.swift @@ -21,6 +21,16 @@ public final class BroadcastConsumer: Sendable { TrackConsumer(try await ffi.subscribeTrack(name: name, subscription: subscription)) } + /// Fetch one complete group by track name and group sequence without holding + /// a live subscription. The group may still be receiving frames. + public func fetchGroup( + name: String, + sequence: UInt64, + options: FetchGroupOptions? = nil + ) async throws -> GroupConsumer { + GroupConsumer(try await ffi.fetchGroup(name: name, sequence: sequence, options: options)) + } + /// Subscribe to a media track, delivering frames in decode order. `container` /// comes from the catalog; `maxLatencyMs` bounds buffering before skipping a GoP. /// `subscription` tunes delivery (priority, ordering, group range); omit for defaults. diff --git a/swift/Sources/Moq/Dynamic.swift b/swift/Sources/Moq/Dynamic.swift index af6594d4ad..6521685032 100644 --- a/swift/Sources/Moq/Dynamic.swift +++ b/swift/Sources/Moq/Dynamic.swift @@ -21,12 +21,75 @@ public final class TrackRequest: Sendable { TrackProducer(try ffi.accept(info: info)) } + /// Create a fetch handler before accepting this requested track. + public func dynamic() throws -> TrackDynamic { + TrackDynamic(try ffi.dynamic()) + } + /// Reject the request with an application error code, failing the subscriber. public func abort(errorCode: Int32) throws { try ffi.abort(errorCode: errorCode) } } +/// A request to produce one uncached group for a fetch consumer. +public final class GroupRequest: Sendable { + let ffi: MoqGroupRequest + + init(_ ffi: MoqGroupRequest) { + self.ffi = ffi + } + + /// The requested group sequence within the track. + public var sequence: UInt64 { + ffi.sequence() + } + + /// The consumer's delivery priority for this fetch. + public var priority: UInt8 { + ffi.priority() + } + + /// Accept the request and return a producer for the group. + public func accept() throws -> GroupProducer { + GroupProducer(try ffi.accept()) + } + + /// Reject the fetch with an application error code. + public func abort(errorCode: Int32) throws { + try ffi.abort(errorCode: errorCode) + } +} + +/// A stream of uncached group requests for one track. +public final class TrackDynamic: AsyncSequence, Sendable { + /// The group request emitted by this sequence. + public typealias Element = GroupRequest + + let ffi: MoqTrackDynamic + + init(_ ffi: MoqTrackDynamic) { + self.ffi = ffi + } + + /// Wait for the next uncached group request. + public func requestedGroup() async throws -> GroupRequest { + GroupRequest(try await ffi.requestedGroup()) + } + + /// Cancel all current and future group request waits. + public func cancel() { + ffi.cancel() + } + + /// Create an iterator that cancels native waits when iteration ends. + public func makeAsyncIterator() -> AsyncThrowingStream.Iterator { + moqStream(cancel: { [ffi] in ffi.cancel() }) { [ffi] in + GroupRequest(try await ffi.requestedGroup()) + }.makeAsyncIterator() + } +} + /// A stream of track requests from subscribers for tracks that are not published /// yet. Iterate directly: `for try await request in dynamic { ... }`. Hold this /// while such requests should be served; the sequence ends (throwing `Closed`) diff --git a/swift/Sources/Moq/Track.swift b/swift/Sources/Moq/Track.swift index 9efd1bb527..7fa2370a03 100644 --- a/swift/Sources/Moq/Track.swift +++ b/swift/Sources/Moq/Track.swift @@ -111,6 +111,11 @@ public final class TrackProducer: Sendable { try await ffi.unused() } + /// Serve fetches for groups that are not currently cached. + public func dynamic() throws -> TrackDynamic { + TrackDynamic(try ffi.dynamic()) + } + /// Append a new group, returning a producer for its frames. public func appendGroup() throws -> GroupProducer { GroupProducer(try ffi.appendGroup()) diff --git a/swift/Tests/MoqTests/SmokeTests.swift b/swift/Tests/MoqTests/SmokeTests.swift index 4091bc0c14..85649656df 100644 --- a/swift/Tests/MoqTests/SmokeTests.swift +++ b/swift/Tests/MoqTests/SmokeTests.swift @@ -38,4 +38,24 @@ final class SmokeTests: XCTestCase { try track.finish() try broadcast.finish() } + + func testBroadcastConsumerFetchesCachedGroup() async throws { + let broadcast = try BroadcastProducer() + let track = try broadcast.publishTrack(name: "events") + let group = try track.appendGroup() + try group.writeFrame(Data("cached".utf8)) + try group.finish() + + let consumer = try broadcast.consume() + let fetched = try await consumer.fetchGroup( + name: "events", + sequence: 0, + options: FetchGroupOptions(priority: 3) + ) + XCTAssertEqual(fetched.sequence, 0) + let frame = try await fetched.readFrame() + XCTAssertEqual(frame, Data("cached".utf8)) + let end = try await fetched.readFrame() + XCTAssertNil(end) + } } From 99494e5f04899bc07227ecdec2d6dd82ac5477a0 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 10 Jul 2026 13:47:56 -0700 Subject: [PATCH 2/2] ci(kt): skip artifact signing on keyless Kotlin builds Fork PRs can't read repo secrets, so the `release-kt-lib.yml` dry-run (`publishToMavenLocal`) ran without a signing key and failed. Two coupled causes: - The job set `ORG_GRADLE_PROJECT_signingInMemoryKey` at job level, so an empty secret left the property present-but-blank and vanniktech tried to sign with an empty key ("Could not read PGP secret key"). - `signAllPublications()` was called unconditionally, registering a *required* sign task that fails ("no configured signatory") even when the key is fully absent. Fix both: scope the signing credentials to the "Publish to Maven Central" step (push/workflow_dispatch only), and gate `signAllPublications()` on a non-blank `signingInMemoryKey`. Keyless local and fork-PR builds now publish unsigned; real releases still sign. This is what the build comment already claimed but never did. Co-Authored-By: Claude Opus 4.8 --- .github/workflows/release-kt-lib.yml | 8 ++++++-- kt/moq/build.gradle.kts | 11 ++++++++--- 2 files changed, 14 insertions(+), 5 deletions(-) diff --git a/.github/workflows/release-kt-lib.yml b/.github/workflows/release-kt-lib.yml index 1979363c83..7e29de517c 100644 --- a/.github/workflows/release-kt-lib.yml +++ b/.github/workflows/release-kt-lib.yml @@ -112,8 +112,10 @@ jobs: # Picked up by com.vanniktech.maven.publish via Gradle project properties. ORG_GRADLE_PROJECT_mavenCentralUsername: ${{ secrets.MAVEN_CENTRAL_USERNAME }} ORG_GRADLE_PROJECT_mavenCentralPassword: ${{ secrets.MAVEN_CENTRAL_PASSWORD }} - ORG_GRADLE_PROJECT_signingInMemoryKey: ${{ secrets.SIGNING_KEY }} - ORG_GRADLE_PROJECT_signingInMemoryKeyPassword: ${{ secrets.SIGNING_PASSWORD }} + # Signing credentials are scoped to the real publish step below, NOT here: + # fork PRs can't read repo secrets, and an empty `signingInMemoryKey` makes + # signAllPublications() attempt (and fail) instead of no-op. The dry-run + # doesn't publish, so it doesn't need to sign. steps: - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6 @@ -150,6 +152,8 @@ jobs: if: github.event_name == 'push' || github.event_name == 'workflow_dispatch' env: MOQ_VERSION: ${{ needs.check-version.outputs.version }} + ORG_GRADLE_PROJECT_signingInMemoryKey: ${{ secrets.SIGNING_KEY }} + ORG_GRADLE_PROJECT_signingInMemoryKeyPassword: ${{ secrets.SIGNING_PASSWORD }} run: | gradle -p kt \ -Pmoq.version="$MOQ_VERSION" \ diff --git a/kt/moq/build.gradle.kts b/kt/moq/build.gradle.kts index acf108cc19..dfb190bcfb 100644 --- a/kt/moq/build.gradle.kts +++ b/kt/moq/build.gradle.kts @@ -18,8 +18,8 @@ // // Publishing uses com.vanniktech.maven.publish; CI runs // `:moq:publishAndReleaseToMavenCentral`. Credentials come from env vars set by -// release-kt-lib.yml (ORG_GRADLE_PROJECT_*). If the signing key isn't set, -// signAllPublications() becomes a no-op so local builds still work. +// release-kt-lib.yml (ORG_GRADLE_PROJECT_*). Signing is only wired up when a key +// is present (see mavenPublishing below), so keyless local and fork-PR builds work. import com.android.build.gradle.LibraryExtension import com.vanniktech.maven.publish.SonatypeHost @@ -128,7 +128,12 @@ if (androidEnabled) { mavenPublishing { publishToMavenCentral(SonatypeHost.CENTRAL_PORTAL, automaticRelease = true) - signAllPublications() + // Only sign when a key is actually configured. signAllPublications() registers a + // *required* sign task, so calling it unconditionally makes publishToMavenLocal + // fail ("no configured signatory") on fork-PR dry-runs, which run without secrets. + if (!providers.gradleProperty("signingInMemoryKey").orNull.isNullOrBlank()) { + signAllPublications() + } coordinates("dev.moq", "moq", version.toString()) pom {