Holding and resetting a stream

Backpressure on a client-initiated stream: hold its receive credit instead of reading faster than the app can keep up, then release it. Separately, abort a server-sent stream the app no longer needs, and check whether one is still in flight.

Code

main.c

#define WIRED_MAIN
#include "wired.h"

/* A server-opened stream, kept open, whose inflight state the client's
 * second round below resets. */
static i64 g_uni  = -1;
static int g_held = 0;

static void on_session(
    void* ctx, wired_wt_session* s, wired_span path, wired_span protocol) {
  (void)ctx;
  (void)path;
  (void)protocol;
  u8  buf[16];
  usz n = wired_wtwire_signal_put(buf, sizeof buf, 0, s->connect_stream_id);
  memcpy(buf + n, "ready", 5);
  g_uni = wired_server_wt_open_uni_stream(s, wired_span_of(buf, n + 5));
}

/* An upload split across two writes: hold receive credit on the first
 * (pausing the peer's send window -- bytes it already has credit for
 * still arrive, nothing new is granted), release once it ends with FIN.
 * The final round's word then exercises reset + inflight on g_uni. */
static void on_stream_data(
    void* ctx, wired_wt_session* s, u64 stream_id, wired_span data, int fin) {
  (void)ctx;
  if (!fin) {
    if (!g_held) {
      wired_server_wt_stream_hold(s, stream_id, 1);
      wired_log_str("held\n");
      g_held = 1;
    }
    return;
  }
  if (g_held) {
    wired_server_wt_stream_hold(s, stream_id, 0);
    wired_log_str("released\n");
    g_held = 0;
  }
  if (wired_span_eq_cstr(data, "abort")) {
    wired_server_wt_stream_reset(s, (u64)g_uni, 1);
    wired_log_str("reset\n");
  }
  wired_dprintf(
      2, "inflight %d\n", wired_server_wt_stream_inflight(s, (u64)g_uni) != 0);
  wired_server_wt_stream_reply(s, stream_id, wired_span_cstr("ok"));
}

int wired_main(int argc, char** argv) {
  /* The same fixed demo identity as the HTTP/3 hello page. */
  static wired_srvboot_demo_keys keys;
  wired_srvboot_id               id;
  wired_srvboot_demo(&id, &keys, 0x50, "guide-wt");
  id.max_datagram_frame_size = 65535; /* WebTransport requires DATAGRAM */

  wired_srvrun_opt opt  = {0};
  opt.incoming_cpu      = -1;
  opt.wt_on_session     = on_session;
  opt.wt_on_stream_data = on_stream_data;

  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 (
  "context"
  "crypto/tls"
  "fmt"
  "io"
  "net/http"
  "os"
  "time"

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

// sendRound opens a fresh bidi stream, writes parts as separate writes (so
// the server sees each as its own reassembly delta), closes it, and
// returns the server's reply.
func sendRound(ctx context.Context, s *webtransport.Session, parts ...string) string {
  str, err := s.OpenStreamSync(ctx)
  if err != nil {
    panic(err)
  }
  for i, p := range parts {
    str.Write([]byte(p))
    if i < len(parts)-1 {
      time.Sleep(100 * time.Millisecond) // let the hold take effect first
    }
  }
  str.Close()
  ack, _ := io.ReadAll(str)
  return string(ack)
}

func main() {
  ctx := context.Background()
  var conn *quic.Conn // kept so main can close it before exiting
  d := webtransport.Transport{
    TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
    QUICConfig:      &quic.Config{EnableDatagrams: true, EnableStreamResetPartialDelivery: true},
    DialAddr: func(ctx context.Context, addr string, t *tls.Config, q *quic.Config) (*quic.Conn, error) {
      c, err := quic.DialAddrEarly(ctx, addr, t, q)
      conn = c
      return c, err
    },
  }
  _, s, err := d.Dial(ctx, os.Args[1]+"/upload", http.Header{})
  if err != nil {
    panic(err)
  }
  defer conn.CloseWithError(0, "")

  fmt.Println("ack1", sendRound(ctx, s, "part1", "part2"))
  fmt.Println("ack2", sendRound(ctx, s, "abort"))
}

Output

$ ./wt-backpressure --port 4433
held
released
inflight 1
reset
inflight 0
conn closed by peer (CONNECTION_CLOSE)

$ go run . https://127.0.0.1:4433
ack1 ok
ack2 ok

How it works

• The first delta of a client upload (fin=0) calls wired_server_wt_stream_hold(s, stream_id, 1): the loop stops raising the advertised MAX_STREAM_DATA, so the peer’s already-granted bytes still arrive but no more credit is handed out.
• The closing delta (fin=1) calls hold(..., 0) to resume, then checks wired_server_wt_stream_inflight(s, g_uni) on the separate server-opened stream from on_session – 1 while it is still open or unacknowledged.
• Sending "abort" instead calls wired_server_wt_stream_reset(s, g_uni, 1), which frees that stream’s send slot at once: inflight reads back 0 right after.
• The client sends each upload as two separate writes before closing, so the server sees two reassembly deltas instead of one – the hold only fires when there is a chunk before the final one.