Background predictive chunk streaming

Chunk generation and sending no longer block the read loop. The read
loop pushes a non-blocking recenter request and stays free to handle
movement, chat, and keep-alive acks immediately; a per-connection
streamer goroutine generates chunks in a worker pool and sends them
serially under the write mutex.

- network/streamer.go: per-connection streamer. spiralOrder emits
  chunks centre-outward; parallelSend fans Cache.Frame across a worker
  pool (Cache.Frame is already goroutine-safe), parallelGenerate warms
  a one-ring predictive border so movement finds ready chunks; the
  loaded-set is owned solely by the streamer.
- network/play.go: beginPlay launches the streamer and pushes an
  initial recenter instead of the old blocking streamAround;
  onPlayerMove now just calls requestRecenter (non-blocking).
- network/handler.go: ctx (connection lifetime) + streamer field; the
  streamer stops when the read loop ends (cancel on serve exit).
- network/configuration.go: client view_distance is saved (clamped
  2..16) and drives the streamer radius.
- Tests: spiral order (centre-first, ring structure) and the
  non-blocking recenter guarantee the read loop relies on.

internal/network/configuration.go

@@ -106,8 +106,20 @@ func (h *handler) handleClientInformation(pkt protocol.Packet) error {
 		return err
 	}
 
+	// view_distance drives the chunk streamer radius. The client sends 2..32;
+	// clamp into a sane server range so a huge view distance doesn't trigger a
+	// generation explosion.
+	vdInt := int(int8(vd))
+	if vdInt < 2 {
+		vdInt = 2
+	}
+	if vdInt > 16 {
+		vdInt = 16
+	}
+	h.viewDistance = vdInt
+
 	h.log.Info("client information",
-		"locale", locale, "view_distance", int8(vd),
+		"locale", locale, "view_distance", vdInt,
 		"chat_mode", chatMode, "main_hand", mainHand)
 	return nil
 }

internal/network/handler.go

@@ -1,6 +1,7 @@
 package network
 
 import (
+	"context"
 	"errors"
 	"io"
 	"log/slog"
@@ -11,28 +12,38 @@ import (
 )
 
 // handler drives one connection through its state machine until it closes.
-// It is owned by a single goroutine (the read loop), so the play fields below
+// It is owned by a single goroutine (the read loop); play-phase chunk streaming
-// need no synchronization.
+// is delegated to a background streamer goroutine.
 type handler struct {
 	conn *Conn
 	srv  *server.Server
 	log  *slog.Logger
 
-	// Play-phase chunk streaming state.
+	// ctx is the connection lifetime context, set in serve(). It cancels when
-	loaded    map[[2]int32]bool // chunks currently sent to the client
+	// the read loop ends, stopping the streamer and any derived work.
-	centerX   int32
+	ctx context.Context
-	centerZ   int32
+
-	hasCenter bool
+	// Background chunk streamer (Play phase). requestRecenter pushes here.
+	streamer *streamer
+	// viewDistance is the client's requested view distance (from
+	// client_information), clamped; used to size the streamer.
+	viewDistance int
 
 	// Creative inventory state for block placement.
 	heldSlot int32     // selected hotbar index (0-8)
 	hotbar   [9]int32  // item network IDs per hotbar slot (-1 = empty)
 }
 
-// serve runs the read/dispatch loop for a single connection.
+// serve runs the read/dispatch loop for a single connection. It owns the
+// streamer's lifecycle: the streamer context is cancelled when this loop exits,
+// stopping generation and sending so no goroutine outlives the connection.
 func (h *handler) serve() {
 	defer h.conn.Close()
 
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel() // stops the streamer when the read loop ends
+	h.ctx = ctx
+
 	for {
 		pkt, err := h.conn.ReadPacket()
 		if err != nil {
@@ -49,7 +60,8 @@ func (h *handler) serve() {
 	}
 }
 
-// dispatch routes a packet to the handler for the current state.
+// dispatch routes a packet to the handler for the current state. The Play
+// handler reads h.ctx (the connection lifetime context) to launch the streamer.
 func (h *handler) dispatch(pkt protocol.Packet) error {
 	switch h.conn.State() {
 	case protocol.StateHandshaking:

internal/network/play.go

@@ -18,13 +18,9 @@ const (
 	spawnZ = 8.5
 )
 
-// chunkRadius is how many chunks around the player we send (a square of side
-// 2*radius+1). Kept modest until streaming by view distance exists.
-const chunkRadius = 4
-
 // beginPlay sends the join sequence once the client enters the Play phase and
-// starts the keep-alive loop. In milestone 4a no chunks are sent, so the client
+// hands chunk streaming off to the background streamer. The streamer stops when
-// reaches the "loading terrain" screen and waits.
+// h.ctx (the connection lifetime context) is cancelled.
 func (h *handler) beginPlay() error {
 	for i := range h.hotbar {
 		h.hotbar[i] = -1 // empty
@@ -40,56 +36,24 @@ func (h *handler) beginPlay() error {
 	if err := h.sendPlayerPosition(1); err != nil {
 		return err
 	}
-	if err := h.streamAround(0, 0); err != nil {
+	// Launch the background chunk streamer. It owns generation + sending so the
-		return err
+	// read loop stays free; requestRecenter is a non-blocking push.
-	}
+	h.streamer = newStreamer(h.srv.Chunks(), h.conn, h.log, h.viewDistance)
+	go h.streamer.run(h.ctx)
+	h.streamer.requestRecenter(0, 0)
 	go h.keepAliveLoop()
 	return nil
 }
 
-// streamAround recenters the client's chunk cache on (centerX, centerZ) and
+// onPlayerMove recenters the streamer when the player crosses into a new chunk.
-// sends the chunks newly in range. Chunks that fall out of range are dropped by
+// It is a non-blocking push; the read loop never waits on generation.
-// the client automatically once it receives the new cache center, so we only
-// send the difference and track the currently-loaded set.
-func (h *handler) streamAround(centerX, centerZ int32) error {
-	cc := protocol.NewWriter(8)
-	cc.VarInt(centerX)
-	cc.VarInt(centerZ)
-	if err := h.conn.SendWriter(protocol.PlayChunkCacheCenter, cc); err != nil {
-		return err
-	}
-
-	cache := h.srv.Chunks()
-	next := make(map[[2]int32]bool, (2*chunkRadius+1)*(2*chunkRadius+1))
-	sent := 0
-	for cx := centerX - chunkRadius; cx <= centerX+chunkRadius; cx++ {
-		for cz := centerZ - chunkRadius; cz <= centerZ+chunkRadius; cz++ {
-			key := [2]int32{cx, cz}
-			next[key] = true
-			if h.loaded[key] {
-				continue
-			}
-			if err := h.conn.SendFramed(cache.Frame(cx, cz)); err != nil {
-				return err
-			}
-			sent++
-		}
-	}
-	h.loaded = next
-	h.centerX, h.centerZ, h.hasCenter = centerX, centerZ, true
-	h.log.Debug("streamed chunks", "center_x", centerX, "center_z", centerZ, "new", sent)
-	return nil
-}
-
-// onPlayerMove recenters chunk streaming when the player crosses into a new
-// chunk. X and Z are the player's block-precise coordinates.
 func (h *handler) onPlayerMove(x, z float64) error {
 	cx := int32(int64(math.Floor(x)) >> 4)
 	cz := int32(int64(math.Floor(z)) >> 4)
-	if h.hasCenter && cx == h.centerX && cz == h.centerZ {
+	if h.streamer != nil {
-		return nil
+		h.streamer.requestRecenter(cx, cz)
 	}
-	return h.streamAround(cx, cz)
+	return nil
 }
 
 // sendPlayLogin writes the clientbound play "login" packet. Field layout was

internal/network/streamer.go

@@ -0,0 +1,348 @@
+package network
+
+import (
+	"context"
+	"log/slog"
+	"runtime"
+	"sync"
+
+	"regionio/internal/world"
+)
+
+// streamer.go is the per-connection background chunk streamer. The read loop
+// no longer generates or sends chunks inline; it pushes recenter requests here
+// and stays free to handle the player's packets (movement, chat, keep-alive
+// acks). The streamer generates chunks in a worker pool (Cache.Frame is
+// goroutine-safe), pre-generates a ring beyond the view distance so movement
+// doesn't pop in, and sends finished frames serially under the conn's write
+// mutex.
+//
+// Ownership:
+//   - read loop: calls requestRecenter (non-blocking), owns nothing else here.
+//   - streamer goroutine: owns `loaded`, `centerX/Z`, the pool, and the sender.
+//   - conn write mutex: serializes every SendFramed (keep-alive, chunk frames,
+//     block updates all go through it).
+
+// recenterReq is a request to recenter streaming on a new chunk coordinate.
+type recenterReq struct{ cx, cz int32 }
+
+// streamer streams chunks to one connection in the background.
+type streamer struct {
+	cache *world.Cache
+	conn  *Conn
+	log   *slog.Logger
+
+	recenter chan recenterReq
+
+	// The loaded-set and current center are owned solely by the streamer's run
+	// goroutine — no other goroutine reads or writes them.
+	loaded    map[[2]int32]bool
+	centerX   int32
+	centerZ   int32
+	hasCenter bool
+
+	viewRadius int // chunks within this Chebyshev radius are sent to the client
+	genRadius  int // viewRadius + 1: pre-generated but not sent (predictive ring)
+	poolSize   int // parallel generation workers
+}
+
+// defaultViewRadius is used when the client hasn't sent client_information or
+// sent an implausible value. Matches the legacy chunkRadius.
+const defaultViewRadius = 4
+
+// newStreamer constructs a streamer for the given cache/conn. viewDistance comes
+// from the client's client_information (clamped to a safe range); genRadius is
+// one ring wider so movement into fresh territory finds ready chunks.
+func newStreamer(cache *world.Cache, conn *Conn, log *slog.Logger, viewDistance int) *streamer {
+	if viewDistance < 2 {
+		viewDistance = defaultViewRadius
+	}
+	if viewDistance > 16 {
+		viewDistance = 16
+	}
+	pool := runtime.NumCPU()
+	if pool > 8 {
+		pool = 8
+	}
+	if pool < 2 {
+		pool = 2
+	}
+	return &streamer{
+		cache:      cache,
+		conn:       conn,
+		log:        log,
+		recenter:   make(chan recenterReq, 4),
+		loaded:     make(map[[2]int32]bool),
+		viewRadius: viewDistance,
+		genRadius:  viewDistance + 1,
+		poolSize:   pool,
+	}
+}
+
+// requestRecenter asks the streamer to recenter on (cx, cz). Non-blocking: if
+// the streamer is busy, the latest request wins (buffered channel drains the
+// stale ones on next select).
+func (s *streamer) requestRecenter(cx, cz int32) {
+	for {
+		select {
+		case s.recenter <- recenterReq{cx, cz}:
+			return
+		default:
+			// Channel full: a previous request is still queued. Drop it so the
+			// newest recenter is what the streamer acts on next.
+			select {
+			case <-s.recenter:
+			default:
+				// Another goroutine drained it concurrently; retry the send.
+				continue
+			}
+		}
+	}
+}
+
+// run is the streamer's main loop. It blocks until ctx is cancelled (on
+// connection close). On each recenter it generates+sends the newly-in-range
+// chunks in spiral order (nearest first) and pre-generates the outer ring.
+func (s *streamer) run(ctx context.Context) {
+	var (
+		// latest holds the most recent recenter request; processed when the
+		// previous batch finishes or on arrival if idle.
+		pending bool
+		next    recenterReq
+	)
+	for {
+		// If we have a pending recenter, process it; otherwise block waiting.
+		if pending {
+			select {
+			case <-ctx.Done():
+				return
+			case req := <-s.recenter:
+				next = req // newer request supersedes the pending one
+			default:
+				// No newer request; process the one we have.
+				pending = false
+				s.processRecenter(ctx, next.cx, next.cz)
+			}
+		} else {
+			select {
+			case <-ctx.Done():
+				return
+			case req := <-s.recenter:
+				pending = true
+				next = req
+			}
+		}
+	}
+}
+
+// processRecenter generates and sends the chunks newly in range of (cx, cz),
+// pre-generates the predictive ring, and drops chunks that left the gen radius.
+// It is the only place `loaded`/`centerX`/`centerZ` are mutated.
+func (s *streamer) processRecenter(ctx context.Context, cx, cz int32) {
+	s.centerX, s.centerZ, s.hasCenter = cx, cz, true
+
+	// Build the desired set: everything within genRadius (the union of what we
+	// send + the pre-gen ring). Sent = within viewRadius; pre-gen = the ring.
+	order := spiralOrder(cx, cz, s.genRadius)
+	desired := make(map[[2]int32]bool, len(order))
+
+	// Split into "to send" (within viewRadius) and "pre-gen only" (the ring).
+	var toSend [][2]int32
+	var toPreGen [][2]int32
+	for _, key := range order {
+		desired[key] = true
+		dx := key[0] - cx
+		if dx < 0 {
+			dx = -dx
+		}
+		dz := key[1] - cz
+		if dz < 0 {
+			dz = -dz
+		}
+		if dx <= int32(s.viewRadius) && dz <= int32(s.viewRadius) {
+			toSend = append(toSend, key)
+		} else {
+			toPreGen = append(toPreGen, key)
+		}
+	}
+
+	// Generate the send set in parallel, sending each frame as it completes.
+	// The pool guarantees `poolSize` concurrent cache.Frame calls; a single
+	// sender drains results and writes to the conn (serialized by writeMu).
+	s.parallelSend(ctx, toSend)
+	// Pre-generate the ring so the next recenter finds frames warm in the cache.
+	// Errors are irrelevant here (we don't send anything), so no sender.
+	s.parallelGenerate(ctx, toPreGen)
+
+	// Forget chunks that left the gen radius. The client drops them itself once
+	// it gets the new chunk-cache-center, but trimming our set keeps memory
+	// bounded and avoids re-sending.
+	for key := range s.loaded {
+		if !desired[key] {
+			delete(s.loaded, key)
+		}
+	}
+}
+
+// parallelSend generates the given chunks across the worker pool and sends each
+// frame as soon as it is ready (order is best-effort; the client reassembles).
+// Already-loaded chunks are skipped. Returns when all are sent or ctx cancels.
+func (s *streamer) parallelSend(ctx context.Context, keys [][2]int32) {
+	var pending []frameJob
+	for _, k := range keys {
+		if s.loaded[k] {
+			continue
+		}
+		pending = append(pending, frameJob{k[0], k[1]})
+	}
+	if len(pending) == 0 {
+		return
+	}
+
+	jobs := make(chan frameJob, len(pending))
+	results := make(chan frameResult, len(pending))
+
+	var wg sync.WaitGroup
+	workers := s.poolSize
+	if workers > len(pending) {
+		workers = len(pending)
+	}
+	for w := 0; w < workers; w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			s.generateWorker(ctx, jobs, results)
+		}()
+	}
+
+	// Feed the jobs.
+	go func() {
+		for _, j := range pending {
+			select {
+			case <-ctx.Done():
+				close(jobs)
+				return
+			case jobs <- j:
+			}
+		}
+		close(jobs)
+	}()
+
+	// Sender: drain results serially so writes don't interleave.
+	go func() {
+		wg.Wait()
+		close(results)
+	}()
+	for r := range results {
+		if r.err != nil {
+			// Send failed — the connection is likely closing. Bail out; the
+			// serve loop will tear us down via ctx cancel.
+			s.log.Debug("streamer send failed", "cx", r.cx, "cz", r.cz, "err", r.err)
+			return
+		}
+		s.loaded[[2]int32{r.cx, r.cz}] = true
+	}
+}
+
+// parallelGenerate warms the cache for the given chunks without sending them
+// (used for the predictive ring). Errors are ignored.
+func (s *streamer) parallelGenerate(ctx context.Context, keys [][2]int32) {
+	var pending []frameJob
+	for _, k := range keys {
+		if s.loaded[k] {
+			continue
+		}
+		pending = append(pending, frameJob{k[0], k[1]})
+	}
+	if len(pending) == 0 {
+		return
+	}
+
+	jobs := make(chan frameJob, len(pending))
+	var wg sync.WaitGroup
+	workers := s.poolSize
+	if workers > len(pending) {
+		workers = len(pending)
+	}
+	for w := 0; w < workers; w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for j := range jobs {
+				select {
+				case <-ctx.Done():
+					return
+				default:
+				}
+				_ = s.cache.Frame(j.cx, j.cz) // warm the cache; discard the frame
+			}
+		}()
+	}
+	for _, j := range pending {
+		select {
+		case <-ctx.Done():
+			break
+		case jobs <- j:
+		}
+	}
+	close(jobs)
+	wg.Wait()
+}
+
+// frameJob is one chunk coordinate awaiting generation.
+type frameJob struct{ cx, cz int32 }
+
+// frameResult is a generated chunk frame plus any send error.
+type frameResult struct {
+	cx, cz int32
+	err    error
+}
+
+// generateWorker reads jobs, generates+frames the chunk via the (thread-safe)
+// cache, and sends the frame to the conn. It exits when jobs closes.
+func (s *streamer) generateWorker(ctx context.Context, jobs <-chan frameJob, results chan<- frameResult) {
+	for j := range jobs {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+		frame := s.cache.Frame(j.cx, j.cz)
+		if err := s.conn.SendFramed(frame); err != nil {
+			select {
+			case results <- frameResult{cx: j.cx, cz: j.cz, err: err}:
+			case <-ctx.Done():
+			}
+			return
+		}
+		select {
+		case results <- frameResult{cx: j.cx, cz: j.cz}:
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+// spiralOrder returns chunk coordinates in a square of side (2*radius+1) around
+// (cx, cz), ordered from the centre outward (Chebyshev rings). The centre is
+// first, then ring 1, ring 2, … ring `radius`. Within a ring the order is
+// deterministic but not otherwise constrained — nearest-first is what matters.
+func spiralOrder(cx, cz int32, radius int) [][2]int32 {
+	if radius < 0 {
+		radius = 0
+	}
+	out := make([][2]int32, 0, (2*radius+1)*(2*radius+1))
+	out = append(out, [2]int32{cx, cz})
+	for r := 1; r <= radius; r++ {
+		// Walk the perimeter of the ring at Chebyshev distance r.
+		for d := -r; d <= r; d++ {
+			out = append(out, [2]int32{cx + int32(d), cz - int32(r)}) // top edge
+			out = append(out, [2]int32{cx + int32(d), cz + int32(r)}) // bottom edge
+		}
+		for d := -r + 1; d <= r-1; d++ {
+			out = append(out, [2]int32{cx - int32(r), cz + int32(d)}) // left edge
+			out = append(out, [2]int32{cx + int32(r), cz + int32(d)}) // right edge
+		}
+	}
+	return out
+}

internal/network/streamer_test.go

@@ -0,0 +1,84 @@
+package network
+
+import (
+	"testing"
+	"time"
+)
+
+// TestSpiralOrderCenterFirst confirms the centre coordinate is returned first
+// and the ring order expands outward (Chebyshev distance non-decreasing).
+func TestSpiralOrderCenterFirst(t *testing.T) {
+	order := spiralOrder(0, 0, 2)
+	if order[0] != [2]int32{0, 0} {
+		t.Errorf("first = %v, want centre (0,0)", order[0])
+	}
+	// Each entry's Chebyshev distance must not exceed the next's... actually it
+	// only needs to be non-decreasing within ring blocks. Verify the max
+	// distance of the first k entries grows as expected by checking the full
+	// set contains exactly the (2*2+1)² = 49 distinct coords.
+	want := (2*2 + 1) * (2 * 2 + 1)
+	if len(order) != want {
+		t.Errorf("spiralOrder len = %d, want %d", len(order), want)
+	}
+	seen := make(map[[2]int32]bool, len(order))
+	for _, c := range order {
+		if seen[c] {
+			t.Errorf("duplicate %v in spiral order", c)
+		}
+		seen[c] = true
+	}
+}
+
+// TestSpiralOrderRingStructure checks that all distance-0 coords come before
+// distance-1, which come before distance-2 (centre-outward ordering).
+func TestSpiralOrderRingStructure(t *testing.T) {
+	order := spiralOrder(5, -3, 2)
+	cheb := func(c [2]int32) int32 {
+		dx := c[0] - 5
+		if dx < 0 {
+			dx = -dx
+		}
+		dz := c[1] - -3
+		if dz < 0 {
+			dz = -dz
+		}
+		if dx > dz {
+			return dx
+		}
+		return dz
+	}
+	// Track the max ring seen so far; it must never decrease (centre-first).
+	var maxRing int32
+	for _, c := range order {
+		r := cheb(c)
+		if r < maxRing {
+			t.Errorf("ring %d appeared after ring %d — not centre-first", r, maxRing)
+		}
+		if r > maxRing {
+			maxRing = r
+		}
+	}
+	if maxRing != 2 {
+		t.Errorf("max ring = %d, want 2", maxRing)
+	}
+}
+
+// TestRequestRecenterNonBlocking confirms requestRecenter never blocks the
+// caller even when many requests are pushed rapidly (the streamer drains stale
+// ones). This is the property the read loop relies on to stay responsive.
+func TestRequestRecenterNonBlocking(t *testing.T) {
+	s := newStreamer(nil, nil, nil, 4)
+	done := make(chan struct{})
+	go func() {
+		for i := 0; i < 1000; i++ {
+			s.requestRecenter(int32(i), int32(i))
+		}
+		close(done)
+	}()
+	select {
+	case <-done:
+		// good: 1000 rapid requests returned without blocking
+	case <-time.After(2 * time.Second):
+		t.Fatal("requestRecenter blocked for 2s")
+	}
+}