Skip to content
wired guide

Priority and GOAWAY

When bandwidth is short, audio should beat video; when a server must go down, its sessions should get a new address and time to leave instead of a dropped connection. This page shows both: two subscriptions with different priorities, whose relay streams get different send classes, and a drain where the hub sends GOAWAY, refuses new requests, ends the granted subscriptions, and closes the session when the grace runs out.

Code

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


/* The hub calls this right after it opens a subscriber stream, with the
 * stream's send class: urgency 4 for every Object stream, then the
 * subscription's Subscriber Priority and the stream's Publisher Priority
 * packed into fine (lower is sent first), then its place inside the
 * subscription. Logging it makes the class visible; the scheduling itself
 * is wired_server_wt_stream_schedule's job. */
static int stream_sched(
    wired_wt_session* s, u64 stream_id, const wired_wt_sched* k) {
  wired_dprintf(
      2, "subscriber stream urgency=%u subscriber=%u publisher=%u\n",
      k->urgency, k->fine >> 8, k->fine & 0xff);
  return wired_server_wt_stream_schedule(s, stream_id, k);
}


static wired_moqt_hub g_hub;


/* Any datagram is the operator's "drain now" trigger in this demo (a real
 * server would call this on SIGTERM or before a deploy). Every session
 * gets GOAWAY pointing at the new URI, with 300 ms to leave (short so the
 * page's client stays well inside its own deadline; the hub still waits
 * WIRED_MOQTRUN_GOAWAY_GRACE_MS past that before it actually closes). */
static void on_datagram(void* ctx, wired_wt_session* s, wired_span data) {
  (void)s;
  (void)data;
  static const char uri[] = "https://relay2.example/moqt";
  wired_moqt_goaway((wired_moqt_hub*)ctx, wired_span_cstr(uri), 300);
  wired_log_str("GOAWAY sent\n");
}


/* The tick enforces the GOAWAY Timeout: PUBLISH_DONE once it passes, the
 * session close on the tick after. */
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-pg");
  id.max_datagram_frame_size = 65535;


  /* stream_sched turns the subscription's priority into the stream's
   * send class; stream_reset and close_session are what the drain uses to
   * end subscriptions and sessions once the GOAWAY Timeout passes. GOAWAY
   * itself goes out on the hub's uni control stream ("moqt-22", 9.2). */
  wired_moqt_io io = wired_moqraw_io();
  io.stream_sched  = stream_sched;
  wired_moqt_init(&g_hub, io);


  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_datagram       = on_datagram,
      .wt_datagram_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"
  "bytes"
  "context"
  "errors"
  "fmt"
  "io"
  "log"
  "os"
  "time"


  "github.com/quic-go/webtransport-go"


  "wired-guide/moqtclient"
)


// publish announces track guide/name with Track Alias alias: PUBLISH
// (0x1D), answered REQUEST_OK (0x7) on its request stream.
func publish(ctx context.Context, s *moqtclient.Session, id byte, name string, alias byte) {
  _, r := s.Request(ctx, 0x1D, append(moqtclient.Track(id, name), alias, 0))
  if t, _ := moqtclient.ReadMsg(r); t != 0x7 {
    log.Fatalf("PUBLISH %s answered with type %#x", name, t)
  }
}


// subscribe sends SUBSCRIBE (0x3) for guide/name with one parameter,
// SUBSCRIBER_PRIORITY (0x20, a single byte: lower is more important), and
// returns the request stream, on which PUBLISH_DONE arrives later.
func subscribe(ctx context.Context, s *moqtclient.Session, id byte, name string, prio byte) *bufio.Reader {
  reply, r := s.Subscribe(ctx, id, name, 1, 0x20, prio)
  if !reply.Ok {
    log.Fatalf("SUBSCRIBE %s refused, error_code=%d", name, reply.ErrCode)
  }
  fmt.Printf("subscriber: SUBSCRIBE %s priority=%d -> SUBSCRIBE_OK\n", name, prio)
  return r
}


// sendGroup starts Group 0 of track alias on a uni stream and leaves the
// stream open, the way a live encoder keeps writing: SUBGROUP_HEADER
// (Type 0x70: FIRST_OBJECT, which the Original Publisher opening a
// subgroup MUST set (SS2.2), Subgroup 0, default Publisher Priority),
// Object ID delta 0, length, payload.
func sendGroup(ctx context.Context, s *moqtclient.Session, alias byte, payload string) {
  st, err := s.OpenUniStreamSync(ctx)
  moqtclient.Check(err)
  _, err = st.Write(append([]byte{0x70, alias, 0, 0, byte(len(payload))}, payload...))
  moqtclient.Check(err)
}


// receive accepts the subscriber's relay stream and prints its first
// Object; the stream stays open and is returned.
func receive(ctx context.Context, s *moqtclient.Session, name string) *bufio.Reader {
  uni, err := s.AcceptUniStream(ctx)
  moqtclient.Check(err)
  r := bufio.NewReader(uni)
  _, group, payload := moqtclient.ReadObject(r)
  fmt.Printf("subscriber: %s group %d %q\n", name, group, payload)
  return r
}


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


  // The publisher announces two tracks.
  pub := moqtclient.Dial(ctx, os.Args[1]+"/moqt")
  defer pub.Conn.CloseWithError(0, "")
  publish(ctx, pub, 0, "video", 1)
  publish(ctx, pub, 2, "audio", 2)


  // The subscriber's session. The hub's GOAWAY arrives later on the hub's
  // uni control stream, the one whose SETUP Dial already read.
  sub := moqtclient.Dial(ctx, os.Args[1]+"/moqt")
  defer sub.Conn.CloseWithError(0, "")


  // Audio matters more than video here: priority 10 against 200. The
  // server log shows the urgency the hub gave each relay stream.
  videoReq := subscribe(ctx, sub, 0, "video", 200)
  audioReq := subscribe(ctx, sub, 2, "audio", 10)
  sendGroup(ctx, pub, 1, "keyframe")
  video := receive(ctx, sub, "video")
  sendGroup(ctx, pub, 2, "frame")
  audio := receive(ctx, sub, "audio")


  // Ask the server to drain (this demo's trigger is any datagram).
  moqtclient.Check(sub.SendDatagram([]byte("drain")))
  typ, body := moqtclient.ReadMsg(sub.Control)
  if typ != 0x10 {
    log.Fatalf("expected GOAWAY, got type %#x", typ)
  }
  br := bufio.NewReader(bytes.NewReader(body))
  uri := make([]byte, moqtclient.Varint(br))
  _, err := io.ReadFull(br, uri)
  moqtclient.Check(err)
  fmt.Printf("subscriber: GOAWAY new_uri=%s timeout=%dms\n", uri, moqtclient.Varint(br))


  // After GOAWAY every new request is refused: REQUEST_ERROR (0x5)
  // GOING_AWAY (0x6). A real client SHOULD NOT start one (SS9.2) and
  // would open its next request on new_uri; this sample sends one on
  // purpose, to show the refusal.
  if reply, _ := sub.Subscribe(ctx, 4, "video"); !reply.Ok {
    fmt.Printf("subscriber: new SUBSCRIBE -> REQUEST_ERROR error_code=%d\n", reply.ErrCode)
  } else {
    log.Fatal("new SUBSCRIBE was accepted after GOAWAY")
  }


  // The subscriptions already granted keep flowing until the timeout.
  // Then the hub resets their relay streams and ends each with
  // PUBLISH_DONE (0xB) GOING_AWAY (0x4) on its request stream.
  for _, st := range []struct {
    name      string
    relay, rq *bufio.Reader
  }{{"video", video, videoReq}, {"audio", audio, audioReq}} {
    _, err := st.relay.ReadByte()
    var se *webtransport.StreamError
    if !errors.As(err, &se) {
      log.Fatalf("%s relay stream: %v", st.name, err)
    }
    fmt.Printf("subscriber: %s relay stream reset error_code=%#x\n", st.name, se.ErrorCode)
    typ, body := moqtclient.ReadMsg(st.rq)
    if typ != 0xB {
      log.Fatalf("expected PUBLISH_DONE, got type %#x", typ)
    }
    fmt.Printf("subscriber: %s PUBLISH_DONE status=%d\n", st.name, moqtclient.Varint(bufio.NewReader(bytes.NewReader(body))))
  }


  // On the next tick the hub closes the session: GOAWAY_TIMEOUT (0x10),
  // since this client left its request streams open.
  _, err = sub.AcceptUniStream(ctx)
  var se *webtransport.SessionError
  if !errors.As(err, &se) {
    log.Fatal(err)
  }
  fmt.Printf("subscriber: session closed error_code=%#x\n", se.ErrorCode)
}

Output

Terminal window
$ ./moqt-priority-goaway --port 4433
subscriber stream urgency=4 subscriber=200 publisher=128
subscriber stream urgency=4 subscriber=10 publisher=128
GOAWAY sent
conn closed by peer (CONNECTION_CLOSE)
conn closed by peer (CONNECTION_CLOSE)
Terminal window
$ go run . https://127.0.0.1:4433
subscriber: SUBSCRIBE video priority=200 -> SUBSCRIBE_OK
subscriber: SUBSCRIBE audio priority=10 -> SUBSCRIBE_OK
subscriber: video group 0 "keyframe"
subscriber: audio group 0 "frame"
subscriber: GOAWAY new_uri=https://relay2.example/moqt timeout=300ms
subscriber: new SUBSCRIBE -> REQUEST_ERROR error_code=6
subscriber: video relay stream reset error_code=0x4
subscriber: video PUBLISH_DONE status=4
subscriber: audio relay stream reset error_code=0x4
subscriber: audio PUBLISH_DONE status=4
subscriber: session closed error_code=0x10

How it works

Section titled “How it works”
  • SUBSCRIBE carries the SUBSCRIBER_PRIORITY parameter (0..255, lower is more important, 128 when absent — draft-ietf-moq-transport-22 9.20.7, 5.1.1). The stream itself carries the Publisher Priority in its SUBGROUP_HEADER. When the hub opens a subscriber’s relay stream it gives the stream a send class (wired_wt_sched): RFC 9218 urgency 4, then subscriber << 8 | publisher as a second key (5.1.2 orders by subscriber priority, then publisher priority), then a place inside the subscription (group order, and a fill stream before the live stream of the same group). Control and request streams keep the transport default urgency 3 and so go first, as 5.1.2 recommends.
  • The class reaches the transport through the hub’s stream_sched io op, here wired_server_wt_stream_schedule wrapped in a logger: audio (priority 10) and video (priority 200) share urgency 4, and the lower second key sends audio bytes before video bytes under pressure. The log shows the class; the scheduling itself is the transport’s job.
  • wired_moqt_goaway(hub, new_uri, timeout_ms) sends GOAWAY (9.2) with the New Session URI on every session’s control stream, which in the moqt-22 sessions of this sample is the server’s unidirectional one (6.6.1 describes the migration). This demo triggers it with any datagram; a real server calls it on SIGTERM or before a deploy. The URI may be up to WIRED_MOQTRUN_GOAWAY_URI_MAX (8192) bytes.
  • After GOAWAY every new request on that session is answered REQUEST_ERROR with error code GOING_AWAY (0x6): the client is expected to take its next request to new_uri. A client SHOULD NOT start new requests after GOAWAY (9.2); this sample sends one SUBSCRIBE on purpose to show the GOING_AWAY answer.
  • The subscriptions already granted keep flowing until the timeout. The hub’s tick (wired_moqt_tick from on_step) enforces it: once timeout_ms passes, each subscription’s relay streams are reset and the subscription ends with PUBLISH_DONE (9.9) status GOING_AWAY (0x4) on its request stream.
  • This client keeps its request streams open past the deadline, so on the next tick the hub closes the session with the GOAWAY_TIMEOUT session code (0x10, WIRED_MOQTRUN_CLOSE_GOAWAY_TIMEOUT). A client that closes its own streams and session in time never hits it.
  • A peer can also ask to leave: WT_DRAIN_SESSION wired to wired_moqt_on_session_draining drains that one session the same way, with an empty URI and a 5000 ms grace.