From a5a29e1e8867800d7670c152f868dc00443b38e5 Mon Sep 17 00:00:00 2001 From: "santi.ferreiro" Date: Mon, 24 Aug 2026 16:05:43 -0300 Subject: [PATCH] fix(push): advertise a valid Largest for the push-mode catalog MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Merge retained installs into the track's largest-location history, send the catalog's PUBLISH after its retained group exists, and re-arm the refresh clock on the catalog's no-demand -> demand edge. Fixes MSF-01 ยง5 catalog bootstrap failing until a reload. --- core/src/facade/publisher.c | 3 ++ service/src/media_sender.c | 31 +++++++++++-------- service/tests/test_media_sender.c | 14 +++++---- .../tests/test_media_sender_idle_refresh.c | 14 +++++---- service/tests/test_media_sender_pacing.c | 13 ++++---- service/tests/test_media_sender_push_cursor.c | 14 +++++---- 6 files changed, 52 insertions(+), 37 deletions(-) diff --git a/core/src/facade/publisher.c b/core/src/facade/publisher.c index 4e69da0..a4909bd 100644 --- a/core/src/facade/publisher.c +++ b/core/src/facade/publisher.c @@ -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; } diff --git a/service/src/media_sender.c b/service/src/media_sender.c index e65aea1..4845fe7 100644 --- a/service/src/media_sender.c +++ b/service/src/media_sender.c @@ -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 @@ -2059,6 +2060,11 @@ 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 @@ -2066,7 +2072,7 @@ static void sender_republish_catalog(moq_media_sender_t *s, uint64_t now_us) * 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; } } @@ -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 @@ -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. */ @@ -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; } diff --git a/service/tests/test_media_sender.c b/service/tests/test_media_sender.c index e517e42..367ca87 100644 --- a/service/tests/test_media_sender.c +++ b/service/tests/test_media_sender.c @@ -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) { diff --git a/service/tests/test_media_sender_idle_refresh.c b/service/tests/test_media_sender_idle_refresh.c index f265a00..94d1cdc 100644 --- a/service/tests/test_media_sender_idle_refresh.c +++ b/service/tests/test_media_sender_idle_refresh.c @@ -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); } diff --git a/service/tests/test_media_sender_pacing.c b/service/tests/test_media_sender_pacing.c index 860bfe7..bdef79d 100644 --- a/service/tests/test_media_sender_pacing.c +++ b/service/tests/test_media_sender_pacing.c @@ -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; @@ -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); @@ -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"); } diff --git a/service/tests/test_media_sender_push_cursor.c b/service/tests/test_media_sender_push_cursor.c index d9a410f..97b0aae 100644 --- a/service/tests/test_media_sender_push_cursor.c +++ b/service/tests/test_media_sender_push_cursor.c @@ -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); }