Skip to content
wired guide

Fetch past Objects

A subscriber that joins late only sees Objects published after it joined. To catch up, it asks for the earlier ones with FETCH, which the hub answers from an Object cache. This page attaches a small cache, has a publisher send Groups 0 to 5, and lets a late viewer subscribe, fill in the past, and join its live subscription seamlessly. It also runs a standalone FETCH.

Code

main.c
#define WIRED_MAIN
#include "wired.h"


/* Room for exactly 4 cached Objects of the 7-byte payloads the client
 * publishes ("frame-N"): each costs MOQCACHE_HDR on top of its payload.
 * The client publishes 6 groups of one Object each, so the oldest 2 are
 * evicted and a FETCH reaching back to them gets an End of Unknown Range. */
#define FRAME_LEN 7
static u8             g_arena[4 * (MOQCACHE_HDR + FRAME_LEN)];
static wired_moqt_hub g_hub;


static void on_step(void* ctx, u64 now_ms) {
  wired_moqt_tick((wired_moqt_hub*)ctx, now_ms);
}


int wired_main(int argc, char** argv) {
  /* The same fixed demo identity as the other MoQT pages, plus QUIC
   * DATAGRAM support, which WebTransport clients ask for. */
  static wired_srvboot_demo_keys keys;
  wired_srvboot_id               id;
  wired_srvboot_demo(&id, &keys, 0x50, "guide-mf");
  id.max_datagram_frame_size = 65535;


  /* Requests arrive on the client's own bidi streams (draft-ietf-moq-
   * transport-22 6.4.2): stream_reply_open answers on them. A FETCH
   * answer, and the fill a SUBSCRIBE asks for (3.4), use open_uni_stream
   * + stream_send rounds, and stream_reset when one is cancelled. */
  wired_moqt_io io = wired_moqraw_io();
  wired_moqt_init(&g_hub, io);


  /* Every whole Object a publisher sends is now kept for FETCH, oldest
   * whole groups evicted first when the arena is full. */
  wired_moqt_cache_attach(&g_hub, g_arena, sizeof g_arena);


  /* The tick retries a FETCH round the transport refused (and gives up
   * on one that stays refused). "moqt-22" selects draft-22. */
  wired_srvrun_opt opt = {
      .incoming_cpu         = -1,
      .wt_protocols         = "moqt-22",
      .on_step              = on_step,
      .on_step_ctx          = &g_hub,
      .wt_on_session        = wired_moqt_on_session,
      .wt_session_ctx       = &g_hub,
      .wt_on_stream_data    = wired_moqt_on_stream_data,
      .wt_stream_data_ctx   = &g_hub,
      .wt_on_stream_reset   = wired_moqt_on_stream_reset,
      .wt_stream_reset_ctx  = &g_hub,
      .wt_on_session_close  = wired_moqt_on_session_close,
      .wt_session_close_ctx = &g_hub,
  };


  u16 port                 = (u16)wired_cliargs_int(argc, argv, "--port", 4433);
  wired_srvrun_handler h   = {0};
  wired_srvrun_obs     obs = {0};
  return wired_server_run_opt(port, &id, h, obs, &opt) ? 0 : 1;
}

Client

client.go
package main


import (
  "bufio"
  "context"
  "fmt"
  "log"
  "os"
  "time"


  "wired-guide/moqtclient"
)


// publishGroup sends Group g of track alias 1 as one Object on its own
// uni stream: SUBGROUP_HEADER (Type 0x70: FIRST_OBJECT -- the Original
// Publisher opening a subgroup MUST set it, SS2.2 -- Subgroup 0, default
// priority, no properties), Object ID delta 0, length, "frame-g".
func publishGroup(ctx context.Context, s *moqtclient.Session, g byte) {
  st, err := s.OpenUniStreamSync(ctx)
  moqtclient.Check(err)
  _, err = st.Write(append([]byte{0x70, 1, g, 0, 7}, fmt.Sprintf("frame-%d", g)...))
  moqtclient.Check(err)
  moqtclient.Check(st.Close())
}


// waitIngested returns once the hub holds Group g. A FETCH of Group g is
// answered REQUEST_ERROR (0x5) INVALID_RANGE (0x11) while g is past the
// track's Largest Object (draft-ietf-moq-transport-22 SS3.2), and
// FETCH_OK (0x18) once it is not, so it is retried every 100 ms, up to
// 3 s.
// Request IDs 2, 4, ... follow the PUBLISH's 0 (each fits one byte).
func waitIngested(ctx context.Context, s *moqtclient.Session, g byte) {
  for id := byte(2); id < 62; id += 2 {
    st, r := s.Request(ctx, 0x16, moqtclient.Fetch(id, "clock", g, g))
    // Our side of this request is done: closing it frees the stream,
    // so the retries never run out of stream credit.
    moqtclient.Check(st.Close())
    typ, _ := moqtclient.ReadMsg(r)
    if typ == 0x18 {
      uni, err := s.AcceptUniStream(ctx) // drain this FETCH's Objects
      moqtclient.Check(err)
      moqtclient.ReadFetch(bufio.NewReader(uni))
      return
    }
    time.Sleep(100 * time.Millisecond)
  }
  log.Fatalf("the hub never took in Group %d", g)
}


// printFetch prints every item of the next fetch data stream and returns
// the last one.
func printFetch(ctx context.Context, s *moqtclient.Session, label string) moqtclient.FetchItem {
  uni, err := s.AcceptUniStream(ctx)
  moqtclient.Check(err)
  rid, items := moqtclient.ReadFetch(bufio.NewReader(uni))
  for _, it := range items {
    switch {
    case it.Unknown:
      fmt.Printf("%s (request %d): end of unknown range up to %d/%d\n", label, rid, it.Group, it.Object)
    case it.TimedOut:
      fmt.Printf("%s (request %d): end of timed-out range up to %d/%d\n", label, rid, it.Group, it.Object)
    case it.Range:
      fmt.Printf("%s (request %d): end of non-existent range up to %d/%d\n", label, rid, it.Group, it.Object)
    default:
      fmt.Printf("%s (request %d): %d/%d %q\n", label, rid, it.Group, it.Object, it.Payload)
    }
  }
  return items[len(items)-1]
}


func main() {
  ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
  defer cancel()


  // The publisher: PUBLISH guide/clock (Request ID 0, Track Alias 1),
  // answered by REQUEST_OK (0x7), then Groups 0..5. The hub caches them
  // in its 4-Object arena, so Groups 0 and 1 are evicted.
  pub := moqtclient.Dial(ctx, os.Args[1]+"/moqt")
  defer pub.Conn.CloseWithError(0, "")
  _, r := pub.Request(ctx, 0x1D, append(moqtclient.Track(0, "clock"), 1, 0))
  if typ, _ := moqtclient.ReadMsg(r); typ != 0x7 {
    log.Fatalf("PUBLISH answered with type %#x", typ)
  }
  for g := byte(0); g < 6; g++ {
    publishGroup(ctx, pub, g)
  }
  // The publisher asks the hub itself whether Group 5 has arrived, so
  // the viewer below joins at a known Largest Object.
  waitIngested(ctx, pub, 5)


  // (1) The viewer joins late with the last 3 Groups, the current one
  // included: SUBSCRIBE (Request ID 0) with LOCATION_FILTER Next Object
  // for the live part and a FILL_PARAMETERS whose LOCATION_FILTER
  // Relative Start 3 asks for Groups 3..5 (Largest Group + 1 - 3 up to
  // the Largest Object). SUBSCRIBE_OK's LARGEST_OBJECT is that Largest
  // Object; the live part starts after it.
  view := moqtclient.Dial(ctx, os.Args[1]+"/moqt")
  defer view.Conn.CloseWithError(0, "")
  reply, _ := view.Subscribe(ctx, 0, "clock", moqtclient.JoinBack(3)...)
  if !reply.Ok || !reply.HasLargest {
    log.Fatalf("SUBSCRIBE failed: %+v", reply)
  }
  fmt.Printf("subscribe (request 0): largest object %d/%d\n", reply.LargestGroup, reply.LargestObject)


  // The fill arrives on a fetch stream of its own whose FETCH_HEADER
  // carries the SUBSCRIBE's Request ID; there is no FETCH_OK.
  last := printFetch(ctx, view, "fill")


  // (2) A FETCH (Request ID 2) of Groups 0..2: FETCH_OK on the request
  // stream with its inclusive End Location, the Objects on a uni stream
  // of their own. The evicted Groups 0 and 1 are asked of the publisher,
  // which does not answer: after the hub's budget they come back as one
  // End of Timed-Out Range.
  _, r = view.Request(ctx, 0x16, moqtclient.Fetch(2, "clock", 0, 2))
  eg, eo := moqtclient.ReadFetchOk(r)
  fmt.Printf("standalone fetch (request 2): FETCH_OK end %d/%d\n", eg, eo)
  printFetch(ctx, view, "standalone fetch")


  // The publisher moves on: Groups 6 and 7 reach the viewer live, each on
  // its own uni stream (one at a time here, since separate streams may
  // arrive in either order).
  var firstLive uint64
  for g := byte(6); g < 8; g++ {
    publishGroup(ctx, pub, g)
    uni, err := view.AcceptUniStream(ctx)
    moqtclient.Check(err)
    _, group, payload := moqtclient.ReadObject(bufio.NewReader(uni))
    fmt.Printf("subscription: %d/0 %q\n", group, payload)
    if g == 6 {
      firstLive = group
    }
  }


  // One Object per Group, so the join is seamless when the first live
  // Group is the one right after the fill's last.
  fmt.Printf("join point: fill ends at %d/%d, live starts at %d/0, seamless=%v\n",
    last.Group, last.Object, firstLive, firstLive == last.Group+1)
}

Output

Terminal window
$ ./moqt-fetch --port 4433
conn closed by peer (CONNECTION_CLOSE)
conn closed by peer (CONNECTION_CLOSE)
Terminal window
$ go run . https://127.0.0.1:4433
subscribe (request 0): largest object 5/0
fill (request 0): 3/0 "frame-3"
fill (request 0): 4/0 "frame-4"
fill (request 0): 5/0 "frame-5"
standalone fetch (request 2): FETCH_OK end 2/0
standalone fetch (request 2): end of timed-out range up to 1/4611686018427387903
standalone fetch (request 2): 2/0 "frame-2"
subscription: 6/0 "frame-6"
subscription: 7/0 "frame-7"
join point: fill ends at 5/0, live starts at 6/0, seamless=true

How it works

Section titled “How it works”
  • The sample negotiates draft-ietf-moq-transport-22 through the WebTransport subprotocol moqt-22 (the server sets opt.wt_protocols). Each side opens a unidirectional control stream and sends SETUP (9.1), and the hub serves requests only after both SETUPs have arrived.
  • wired_moqt_cache_attach gives the hub an arena for FETCH (draft-ietf-moq-transport-22 9.11): every whole Object a client PUBLISHes is copied into it. Each Object costs MOQCACHE_HDR bytes on top of its payload, so this arena holds exactly 4 of the 7-byte frame-N Objects. Publishing 6 Groups evicts the oldest whole Groups, 0 and 1.
  • The hub’s own blob and live tracks are not cached: only tracks a client PUBLISHes are, so the Go client opens two sessions, a publisher and a viewer. Each request (PUBLISH, SUBSCRIBE, FETCH) goes on a fresh bidi stream the client opens (6.4.2), and the hub answers on that stream through stream_reply_open.
  • The viewer SUBSCRIBEs with the LOCATION_FILTER parameter (9.20.9) set to Next Object (type 0x05, called Largest Object in draft-19): only Objects after the current Largest arrive live. SUBSCRIBE_OK reports that Largest (LARGEST_OBJECT, 9.20.17, 5/0).
  • To catch up, the same SUBSCRIBE carries FILL_PARAMETERS (0x23, 9.20.15) whose inner LOCATION_FILTER is Relative Start N (type 0x01, here N = 3): the last N Groups, the current one included, starting at {Largest.Group + 1 - N, 0} (9.20.9; 3.4, 3.5). draft-22 has no Joining FETCH: the fill rides on the subscription. The hub sends it on a server-opened unidirectional stream whose FETCH_HEADER (11.4.1) carries the SUBSCRIBE’s own Request ID, and there is no FETCH_OK. Here it delivers Groups 3 to 5 (fill (request 0): 3/0 through 5/0). The first live Object is 6/0, so the fill and the subscription meet with no gap and no duplicate.
  • A standalone FETCH of Groups 0 to 2 has the body Request ID, Track Namespace, Track Name, Parameters; the range is the LOCATION_FILTER parameter of type 0x03 (Absolute Start, Group End: start 0/0, End Group Delta 2), with no Fetch Type field. FETCH_OK (9.12) comes on its request stream, with End Location 2/0 (FETCH_OK end 2/0). The End Location is inclusive in draft-22; in draft-19 it was one past the last Object. The Objects arrive on a new uni stream (FETCH_HEADER, then fetch Objects). The cache no longer holds the evicted Groups 0 and 1, so the hub asks the track’s publisher for them with a FETCH of its own (3.2.3). This sample’s publisher does not answer FETCH: once the hub’s budget (WIRED_MOQTRUN_UPFETCH_BUDGET_MS, 2 s, or a shorter FILL_TIMEOUT) has passed, the hub cancels that request and reports the two Groups as one End of Timed-Out Range (0x20C) up to 1/4611686018427387903 (2^62−1, the hub’s stand-in for “the rest of Group 1”; draft-22 Object IDs can go up to 2^64−1), then Group 2 from the cache. A publisher that answers would have its Objects relayed here instead. A draft-19 viewer, which has no 0x20C, gets the same gap as End of Unknown Range (0x10C).
  • The open_uni_stream, stream_send and stream_reset io entries carry the fetch and fill streams: they stay open across rounds, and the hub’s tick (wired_moqt_tick from on_step) retries a round the transport refused.
  • Before the viewer joins, the publisher checks that the hub has taken in Group 5, using FETCH itself: a standalone FETCH of Group 5 is answered REQUEST_ERROR INVALID_RANGE while it is past the track’s Largest Object, and FETCH_OK once the hub has it. The client retries every 100 ms (up to 3 s), closing its side of each request stream so the retries never run out of stream credit. The viewer therefore always joins at Largest 5/0.