Skip to content

fix(push): advertise a valid Largest for the push-mode catalog #11

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
santi-ferreiro wants to merge 1 commit into from
Closed

fix(push): advertise a valid Largest for the push-mode catalog #11

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions core/src/facade/publisher.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1851,6 +1851,9 @@ static moq_result_t track_install_retained(moq_pub_track_t *t, uint64_t group_id
t->retained_group_id = group_id;
t->has_retained = true;
t->wrote_object = false; /* retained group is now the latest published loc */
if (retained_can_advertise_largest(t))
track_hist_merge(t->hist, group_id,
pub_track_retained_last_object_id(t));
return MOQ_OK;
}

Expand Down
31 changes: 18 additions & 13 deletions service/src/media_sender.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -413,6 +413,7 @@ struct moq_media_sender {
* on every generation commit so a real mutation resets the cadence. */
uint64_t catalog_refresh_interval_us;
uint64_t catalog_refresh_deadline_us;
bool catalog_had_demand;
/* Cached next service-deadline for the managed-adapter wake query, recomputed
* at the end of every sender_hook (where facade access is legal) and read by
* sender_next_deadline_us as a pure scalar under s->mu -- so the deadline the
Expand Down Expand Up @@ -2059,14 +2060,19 @@ static void sender_republish_catalog(moq_media_sender_t *s, uint64_t now_us)
(void)sender_stage_generation(s); /* stages, no-op dedups, or fatal */
if (s->fatal) { pthread_mutex_unlock(&s->mu); return; }
}
bool cat_demand = sender_track_demand(s, s->catalog_track);
if (cat_demand != s->catalog_had_demand) {
s->catalog_had_demand = cat_demand;
if (cat_demand) sender_arm_refresh(s, now_us);
}
/* No mutation generation this cycle (dedup, or nothing was dirty): if a
* periodic refresh is due AND the catalog has demand, stage exactly one
* independent refresh. No demand -> no refresh / no group advance. Never
* a redundant second generation in one cycle (a mutation above leaves
* pending_obj_count > 0, skipping this). */
if (s->pending_obj_count == 0 &&
sender_refresh_due(s, now_us) &&
sender_track_demand(s, s->catalog_track)) {
cat_demand) {
(void)sender_stage_refresh(s);
if (s->fatal) { pthread_mutex_unlock(&s->mu); return; }
}
Expand Down Expand Up @@ -2410,17 +2416,6 @@ static void sender_hook(moq_endpoint_t *ep, moq_session_t *session,
sender_set_fatal(s, MOQ_MEDIA_SENDER_FATAL_SETUP_FAILED);
return;
}
/* PUBLISH the catalog track (media tracks are published in
* sender_add_pub_track below). */
if (s->publish_tracks) {
moq_pub_publish_cfg_t cpcfg;
moq_pub_publish_cfg_init(&cpcfg);
if (moq_pub_publish_track(s->pub, s->catalog_track->pub_track,
&cpcfg, now_us) != MOQ_OK) {
sender_set_fatal(s, MOQ_MEDIA_SENDER_FATAL_SETUP_FAILED);
return;
}
}

/* Walk a snapshot so a concurrent app-thread realloc of s->tracks cannot
* free the vector mid-walk; the per-track registration below takes s->mu
Expand Down Expand Up @@ -2489,6 +2484,16 @@ static void sender_hook(moq_endpoint_t *ep, moq_session_t *session,
s->published_catalog = json;
s->catalog_group = 0;
s->catalog_published = true;

if (s->publish_tracks) {
moq_pub_publish_cfg_t cpcfg;
moq_pub_publish_cfg_init(&cpcfg);
if (moq_pub_publish_track(s->pub, s->catalog_track->pub_track,
&cpcfg, now_us) != MOQ_OK) {
sender_set_fatal(s, MOQ_MEDIA_SENDER_FATAL_SETUP_FAILED);
return;
}
}
/* Start the automatic-refresh clock from the initial catalog install. */
sender_arm_refresh(s, now_us);
/* Record the initial catalog's track set as the published baseline. */
Expand Down Expand Up @@ -2533,7 +2538,7 @@ static void sender_hook(moq_endpoint_t *ep, moq_session_t *session,
ocfg.end_of_group = true;
moq_result_t wrc = moq_pub_write_object_ex(
s->pub, s->catalog_track->pub_track, &ocfg, now_us);
if (wrc == MOQ_OK)
if (wrc == MOQ_OK && sender_track_demand(s, s->catalog_track))
s->live_catalog_sent = true;
}

Expand Down
14 changes: 8 additions & 6 deletions service/tests/test_media_sender.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -206,14 +206,16 @@ static int server_pump(moq_pq_threaded_t *t, moq_pq_threaded_lane_t *lane,
* track's publication handle so its live generations (delivered
* WITHOUT any SUBSCRIBE) can be captured below. */
const moq_publish_request_event_t *pr = &ev.u.publish_request;
moq_accept_publish_cfg_t acc;
moq_accept_publish_cfg_init(&acc);
if (moq_session_accept_publish(session, pr->pub, &acc,
now_us) == MOQ_OK &&
const bool is_catalog =
pr->track_name.len == strlen(MOQ_MSF_CATALOG_TRACK_NAME) &&
memcmp(pr->track_name.data, MOQ_MSF_CATALOG_TRACK_NAME,
pr->track_name.len) == 0) {
st->pub_cat = pr->pub;
pr->track_name.len) == 0;
const moq_publication_t cand = pr->pub;
moq_accept_publish_cfg_t acc;
moq_accept_publish_cfg_init(&acc);
if (moq_session_accept_publish(session, cand, &acc,
now_us) == MOQ_OK && is_catalog) {
st->pub_cat = cand;
st->pub_cat_ok = true;
}
} else if (ev.kind == MOQ_EVENT_OBJECT_RECEIVED) {
Expand Down
14 changes: 8 additions & 6 deletions service/tests/test_media_sender_idle_refresh.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,15 +55,17 @@ static int psrv_pump(moq_pq_threaded_t *t, moq_pq_threaded_lane_t *lane,
ev.u.namespace_published.ann, &acc, now_us);
} else if (ev.kind == MOQ_EVENT_PUBLISH_REQUEST) {
const moq_publish_request_event_t *pr = &ev.u.publish_request;
moq_accept_publish_cfg_t acc;
moq_accept_publish_cfg_init(&acc);
if (moq_session_accept_publish(session, pr->pub, &acc,
now_us) == MOQ_OK &&
const bool is_catalog =
pr->track_name.len == strlen(MOQ_MSF_CATALOG_TRACK_NAME) &&
memcmp(pr->track_name.data, MOQ_MSF_CATALOG_TRACK_NAME,
pr->track_name.len) == 0) {
pr->track_name.len) == 0;
const moq_publication_t cand = pr->pub;
moq_accept_publish_cfg_t acc;
moq_accept_publish_cfg_init(&acc);
if (moq_session_accept_publish(session, cand, &acc,
now_us) == MOQ_OK && is_catalog) {
pthread_mutex_lock(&ps->mu);
ps->pub_cat = pr->pub;
ps->pub_cat = cand;
ps->pub_cat_ok = true;
pthread_mutex_unlock(&ps->mu);
}
Expand Down
13 changes: 7 additions & 6 deletions service/tests/test_media_sender_pacing.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -233,6 +233,7 @@ typedef struct {
bool accepted; /* did this fixture answer it? */
uint64_t pub_opaque; /* exact handle, DERIVED pre-ingress */
uint64_t track_alias; /* exact alias, DERIVED pre-ingress */
bool has_largest; /* catalog only: PUBLISH follows the retained group */
} pubreq_exp_t;
typedef struct {
int pass;
Expand Down Expand Up @@ -988,7 +989,7 @@ static void expect_pub_requests(fix_t *fix, const pubreq_exp_t *exp, int n,
MOQ_TEST_CHECK_EQ_SIZE(q->token_count, 0);
MOQ_TEST_CHECK_EQ_SIZE(q->props_len, 0);
MOQ_TEST_CHECK(!q->dynamic_groups);
MOQ_TEST_CHECK(!q->has_largest);
MOQ_TEST_CHECK(q->has_largest == exp[e].has_largest);
MOQ_TEST_CHECK_EQ_U64(q->largest_group, 0);
MOQ_TEST_CHECK_EQ_U64(q->largest_object, 0);
MOQ_TEST_CHECK(!q->has_expires);
Expand Down Expand Up @@ -3462,13 +3463,13 @@ static void test_reset_partial_fanout(void)
* EXACTLY the video one, so only that track fans out. */
{
/* Source-declared request ORDER decides which derived handle/alias
* belongs to which track: catalog first, then the media snapshot in
* add order (v, a). */
* belongs to which track: the media snapshot in add order (v, a), then
* the catalog, whose PUBLISH is sent after its retained group exists. */
const pubreq_exp_t preq[3] = {
{ "v", true, fix.exp_pub_handle[0], fix.exp_pub_alias[0] },
{ "a", false, fix.exp_pub_handle[1], fix.exp_pub_alias[1] },
{ MOQ_MSF_CATALOG_TRACK_NAME, false,
fix.exp_pub_handle[0], fix.exp_pub_alias[0] },
{ "v", true, fix.exp_pub_handle[1], fix.exp_pub_alias[1] },
{ "a", false, fix.exp_pub_handle[2], fix.exp_pub_alias[2] },
fix.exp_pub_handle[2], fix.exp_pub_alias[2], true },
};
expect_pub_requests(&fix, preq, 3, "fanout-publish-requests");
}
Expand Down
14 changes: 8 additions & 6 deletions service/tests/test_media_sender_push_cursor.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,15 +67,17 @@ static int psrv_pump(moq_pq_threaded_t *t, moq_pq_threaded_lane_t *lane,
atomic_store(&ps->ns_accepted, true);
} else if (ev.kind == MOQ_EVENT_PUBLISH_REQUEST) {
const moq_publish_request_event_t *pr = &ev.u.publish_request;
moq_accept_publish_cfg_t acc;
moq_accept_publish_cfg_init(&acc);
if (moq_session_accept_publish(session, pr->pub, &acc,
now_us) == MOQ_OK &&
const bool is_catalog =
pr->track_name.len == strlen(MOQ_MSF_CATALOG_TRACK_NAME) &&
memcmp(pr->track_name.data, MOQ_MSF_CATALOG_TRACK_NAME,
pr->track_name.len) == 0) {
pr->track_name.len) == 0;
const moq_publication_t cand = pr->pub;
moq_accept_publish_cfg_t acc;
moq_accept_publish_cfg_init(&acc);
if (moq_session_accept_publish(session, cand, &acc,
now_us) == MOQ_OK && is_catalog) {
pthread_mutex_lock(&ps->mu);
ps->pub_cat = pr->pub;
ps->pub_cat = cand;
ps->pub_cat_ok = true;
pthread_mutex_unlock(&ps->mu);
}
Expand Down