feat: a seam for shards, a quicker hot path, metrics, deploys without a storm, and a load harness

CLAUDE.md

@@ -44,6 +44,20 @@ maintains it.
   so it cannot make data look fresher; keep it that way.
 - **Import only `go/protocol` from `srcfl/ftw`.** `scripts/check-imports.sh`
   fails on anything else.
+- **A home lives on one shard** (`internal/shard`). An instance that does not
+  hold a home names the host that does (4421, or 421 with `Ftw-Shard`)
+  before it reads message 1 or touches the registry. Nothing about a home
+  crosses between instances: no shared pub/sub, no shared storage. With no
+  `-shards` file one instance holds every home. `/mcp` and `/oauth` stay on
+  `cloud.ftw.energy` and forward to the home's shard once there is a second
+  one, because other parties' clients follow no redirect.
+- **Counts name nobody.** `/metrics` listens on loopback only, and it and
+  the minute's log line sum over all homes: no home id, device or address.
+- **A deploy drains.** On SIGTERM every socket, handshakes included, hears
+  4503 with a spread before the process exits, and no drained box starts
+  `box.unreachable`. A part that holds sockets is a `drainer` in
+  `cmd/ftwcloudd`. Every close the service sends waits for the client's
+  answer (`internal/wsclose`), so the client reads the code, not a reset.
 
 ## Work
 
@@ -54,20 +68,25 @@ maintains it.
   and the service is over it. The one-door plan gives each part a budget of
   its own, and the numbers are Fredrik's call. Until he sets them, a PR that
   grows a part says by how much. Lines of Go without tests, at
-  srcfl/ftw-cloud#15:
+  srcfl/ftw-cloud#17 (#15 in brackets):
 
   | Part | Lines |
   |---|---|
-  | `internal/appsession` (app session, one-shot read) | 1,020 |
-  | `internal/store` | 807 |
-  | `internal/uplink` | 719 |
-  | `internal/registry` | 694 |
-  | `cmd/ftwcloudd` | 553 |
-  | `internal/push` | 542 |
-  | `internal/apns` | 299 |
-  | `internal/server` | 205 |
-  | `internal/token` | 170 |
-  | All | 5,009 |
+  | `internal/appsession` (app session, one-shot read) | 1,116 (1,020) |
+  | `internal/push` | 1,029 (542) |
+  | `internal/uplink` | 938 (719) |
+  | `internal/store` | 854 (807) |
+  | `internal/registry` | 809 (694) |
+  | `cmd/ftwcloudd` | 701 (553) |
+  | `internal/metrics` | 448 (new) |
+  | `internal/apns` | 299 (299) |
+  | `internal/token` | 214 (170) |
+  | `internal/server` | 199 (205) |
+  | `internal/shard` | 188 (new) |
+  | `internal/wsclose` | 109 (new) |
+  | All | 6,904 (5,009) |
+
+  `cmd/ftwload`, the load harness, is a tool and not the service (1,938).
 - Nothing is deployed from here without the owner. `deploy/` describes the
   running host; it does not run itself.
 - Keep task breakdowns and agent notes out of the repository.

README.md

@@ -30,6 +30,14 @@ to it yet.
   and when an away box's alert is due. A box whose owner turns the readable
   copy off gets that copy deleted and a signed receipt (`purged {what:
   "copy"}`); a box whose link ends gets its home deleted and a receipt.
+- **Room to grow.** Each home lives on one shard; an instance that does not
+  hold a home names the host that does, and clients follow once
+  (`internal/shard`, protocol "Shards and restarts"). With no `-shards` file
+  one instance holds every home. A deploy closes every socket with 4503 and
+  a spread, so boxes and phones come back over seconds rather than at once.
+  `/metrics` on loopback and one log line a minute count what the service
+  does without naming a home. `deploy/README.md` has the load numbers and
+  when to add a shard.
 - **A restart forgets no one.** At start the service reads every signed
   roster back from the registry and checks it with the home's key, so a
   home's phones are admitted before its box reconnects, and the wait for
@@ -206,8 +214,14 @@ go test -run XXX -fuzz FuzzPhoneMessage -fuzztime 30s ./internal/appsession
 go test -run TestTheSnapshotArrivesFast -v ./internal/appsession   # loopback p50/p95
 go test -run XXX -bench BenchmarkSnapshot ./internal/appsession
 bash scripts/check-imports.sh
+go build -o /tmp/ftwload/ftwcloudd ./cmd/ftwcloudd   # then, briefly and on a quiet machine:
+go run ./cmd/ftwload -ftwcloudd /tmp/ftwload/ftwcloudd -homes 1000 -phones 1 -duration 45s
 ```
 
+`cmd/ftwload` starts its own `ftwcloudd`, drives simulated boxes and phones
+against it, prints latencies (its own and the service's from `/metrics`),
+CPU and memory, and exits by itself.
+
 The wire types come from `github.com/srcfl/ftw/go/protocol`, Apache-2.0,
 pinned to a commit in `go.mod`. A wire change is a pair of pull requests:
 first in `srcfl/ftw` (`go/protocol` and its golden vectors), then here, moving

cmd/ftwcloudd/main.go

@@ -4,6 +4,7 @@
 //
 //	ftwcloudd keygen -out service.key          make the service key; prints its public half
 //	ftwcloudd keygen -noise -out noise.key     make the Noise key phones dial; prints its public half
+//	ftwcloudd metrics                          print the running service's counts, from inside its container
 //	ftwcloudd [flags]                          serve
 package main
 
@@ -23,19 +24,23 @@ import (
 	"io"
 	"log"
 	"log/slog"
+	"net"
 	"net/http"
 	"os"
 	"os/signal"
 	"path/filepath"
+	"strconv"
 	"strings"
 	"syscall"
 	"time"
 
 	"github.com/srcfl/ftw-cloud/internal/apns"
 	"github.com/srcfl/ftw-cloud/internal/appsession"
+	"github.com/srcfl/ftw-cloud/internal/metrics"
 	"github.com/srcfl/ftw-cloud/internal/push"
 	"github.com/srcfl/ftw-cloud/internal/registry"
 	"github.com/srcfl/ftw-cloud/internal/server"
+	"github.com/srcfl/ftw-cloud/internal/shard"
 	"github.com/srcfl/ftw-cloud/internal/store"
 	"github.com/srcfl/ftw-cloud/internal/token"
 	"github.com/srcfl/ftw-cloud/internal/uplink"
@@ -52,6 +57,13 @@ func main() {
 		}
 		return
 	}
+	if len(os.Args) > 1 && os.Args[1] == "metrics" {
+		if err := printMetrics(os.Args[2:]); err != nil {
+			fmt.Fprintln(os.Stderr, err)
+			os.Exit(1)
+		}
+		return
+	}
 	if err := serve(os.Args[1:]); err != nil {
 		slog.Error("ftwcloudd stopped", "err", err)
 		os.Exit(1)
@@ -73,6 +85,10 @@ func serve(args []string) error {
 	apnsWatch := fs.String("apns-watch-topic", "energy.ftw.app.watchkitapp", "the watch app's bundle id")
 	devPush := fs.String("dev-push-listen", "", "loopback address for registering a token by hand (development builds only)")
 	devCapture := fs.String("dev-apns-capture", "", "write each alert to this directory instead of sending it to Apple (development builds only)")
+	shardsPath := fs.String("shards", "", "shard directory, JSON; without it this instance holds every home")
+	metricsAddr := fs.String("metrics-listen", defaultMetricsAddr, "loopback address for /metrics; empty turns it off")
+	restartRate := fs.Int("restart-rate", defaultRestartRate, "sockets a second the next instance can take after a restart; sets the spread of the redial")
+	restartMin := fs.Duration("restart-spread-min", 15*time.Second, "the shortest spread: longer than the port stays closed while the next container starts")
 	_ = fs.Parse(args)
 
 	harden()
@@ -90,6 +106,15 @@ func serve(args []string) error {
 	} else if err != nil {
 		return err
 	}
+	shards := shard.Single(*domain)
+	if *shardsPath != "" {
+		if shards, err = shard.Load(*shardsPath); err != nil {
+			return err
+		}
+		if shards.Host() != *domain {
+			return fmt.Errorf("shards: this instance is %s at %s, but -domain is %s", shards.Self(), shards.Host(), *domain)
+		}
+	}
 	reg, err := registry.Open(filepath.Join(*stateDir, "registry.db"))
 	if err != nil {
 		return err
@@ -99,15 +124,18 @@ func serve(args []string) error {
 	st := store.New(nil)
 	st.UseSuspensions(reg.Suspended)
 	up := uplink.New(st, reg, key, log)
+	up.Shards = shards
 	// A restart is a deploy, and commands ride through here: phones are
 	// admitted by the rosters their boxes signed before, without waiting
 	// for every box to reconnect.
+	restoreStart := time.Now()
 	if n, err := up.Restore(); err != nil {
 		return err
 	} else if n > 0 {
-		log.Info("rosters restored from the registry", "homes", n)
+		log.Info("rosters restored from the registry", "homes", n, "took", time.Since(restoreStart).Round(time.Millisecond))
 	}
 	tokens := token.New(st, reg, log)
+	tokens.Shards = shards
 	srv := &server.Server{Store: st, Registry: reg, Uplink: up, Tokens: tokens, Log: log}
 
 	// Alerts: the box's signed events to its phones through Apple, and
@@ -129,6 +157,7 @@ func serve(args []string) error {
 	}
 	pushSvc := push.New(reg, sender, log)
 	pushSvc.WatchApp = *apnsWatch
+	pushSvc.Owns = shards.Ours
 	up.OnHeld = pushSvc.Held
 	up.OnRoster = func(home string, r *cloudwire.Roster) {
 		pushSvc.Roster(home, r)
@@ -143,6 +172,7 @@ func serve(args []string) error {
 		srv.App.Push = pushSvc
 		srv.App.Tokens = tokens
 		srv.App.Tunnel = up
+		srv.App.Shards = shards
 	}
 	app := srv.App
 	up.OnConnect = func(home string, conn uint64) {
@@ -184,10 +214,17 @@ func serve(args []string) error {
 		ErrorLog: log0(),
 	}
 
+	metrics.Default.GaugeFunc("ftwcloud_homes", "Homes held in memory.", func() int64 { return int64(st.Homes()) })
+	metrics.Default.GaugeFunc("ftwcloud_homes_connected", "Homes whose box is connected.", func() int64 { return int64(up.Boxes()) })
+	metrics.Runtime()
+
 	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
 	defer stop()
 	go srv.Upkeep(ctx, time.Minute)
-	go pushSvc.Run(ctx)
+	// Push outlives the signal until the queued alerts are out.
+	pushCtx, stopPush := context.WithCancel(context.Background())
+	defer stopPush()
+	go pushSvc.Run(pushCtx)
 	if err := pushSvc.Resume(); err != nil {
 		log.Warn("could not restart the dead man's switch", "err", err)
 	}
@@ -196,22 +233,133 @@ func serve(args []string) error {
 			return err
 		}
 	}
+	if *metricsAddr != "" {
+		if err := serveMetrics(ctx, *metricsAddr); err != nil {
+			return err
+		}
+	}
+	go metrics.Default.Summarise(ctx, os.Stdout, shards.Self(), time.Minute)
+
+	// A deploy: stop accepting, tell every box and phone to come back within
+	// a spread the next instance can take, let the queued alerts go, then
+	// exit. A box closed this way starts no box.unreachable. Every part
+	// that holds sockets is a drainer, so an agent's sessions are closed and
+	// counted in the spread too once that part adds itself here.
+	drainers := []drainer{up}
+	if app != nil {
+		drainers = append(drainers, app)
+	}
+	drained := make(chan struct{})
 	go func() {
+		defer close(drained)
 		<-ctx.Done()
 		shutdown, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
-		_ = httpSrv.Shutdown(shutdown)
+		go func() { _ = httpSrv.Shutdown(shutdown) }()
+		sockets := 0
+		for _, d := range drainers {
+			sockets += d.Sockets()
+		}
+		spread := restartSpread(sockets, *restartRate, *restartMin)
+		reason := strconv.FormatInt(spread.Milliseconds(), 10)
+		for _, d := range drainers {
+			d.Drain(reason)
+		}
+		// Each socket waits for its client's answer to the close, so the
+		// client reads 4503 and the spread rather than a reset.
+		left := 0
+		deadline := time.Now().Add(drainWait)
+		for _, d := range drainers {
+			left += d.WaitClosed(time.Until(deadline))
+		}
+		log.Info("drained for a restart", "sockets", sockets, "still_open", left, "spread", spread)
+		if left := pushSvc.Flush(3 * time.Second); left > 0 {
+			log.Warn("alerts queued or waiting to go again at exit; they are lost", "count", left)
+		}
+		stopPush()
 	}()
 
 	pub, _ := cloudwire.PublicKeyBytesOf(&key.PublicKey)
-	log.Info("ftwcloudd listening", "addr", *listen, "service_key", base64.RawURLEncoding.EncodeToString(pub),
-		"noise_key", base64.RawURLEncoding.EncodeToString(noise.Public()))
-	if err := httpSrv.ListenAndServeTLS("", ""); err != nil && !errors.Is(err, http.ErrServerClosed) {
+	log.Info("ftwcloudd listening", "addr", *listen, "shard", shards.Self(), "shards", shards.Shards(),
+		"service_key", base64.RawURLEncoding.EncodeToString(pub), "noise_key", base64.RawURLEncoding.EncodeToString(noise.Public()))
+	err = httpSrv.ListenAndServeTLS("", "")
+	if err != nil && !errors.Is(err, http.ErrServerClosed) {
 		return err
 	}
+	<-drained
 	return nil
 }
 
+// drainer is a part that holds sockets and closes them before a restart.
+type drainer interface {
+	// Sockets is how many sockets it holds, handshakes included.
+	Sockets() int
+	// Drain closes them with 4503 and reason, and turns away new ones.
+	Drain(reason string)
+	// WaitClosed waits at most d for their clients' answers and returns how
+	// many are still open.
+	WaitClosed(d time.Duration) int
+}
+
+// drainWait bounds the wait for the clients' answers to the drain's close.
+const drainWait = 2 * time.Second
+
+// defaultRestartRate is how many sockets a second one instance takes on
+// the host, with room to spare: deploy/README.md, "Load", has the numbers.
+const defaultRestartRate = 500
+
+// restartSpread is how long sockets wait at most before they redial: their
+// number at rate a second, at least floor, which covers the next instance's
+// start, and at most cloudwire.RestartSpreadMaxMs.
+func restartSpread(sockets, rate int, floor time.Duration) time.Duration {
+	if rate <= 0 {
+		rate = defaultRestartRate
+	}
+	d := time.Duration(sockets) * time.Second / time.Duration(rate)
+	return min(max(d, floor), time.Duration(cloudwire.RestartSpreadMaxMs)*time.Millisecond)
+}
+
+// defaultMetricsAddr is inside the container only; an operator reads it
+// with docker exec and ftwcloudd metrics.
+const defaultMetricsAddr = "127.0.0.1:9464"
+
+// serveMetrics answers /metrics on a loopback address until ctx ends.
+func serveMetrics(ctx context.Context, addr string) error {
+	host, _, err := net.SplitHostPort(addr)
+	if err != nil {
+		return fmt.Errorf("metrics-listen: %w", err)
+	}
+	if ip := net.ParseIP(host); host != "localhost" && (ip == nil || !ip.IsLoopback()) {
+		return fmt.Errorf("metrics-listen: %s is not a loopback address", addr)
+	}
+	ln, err := net.Listen("tcp", addr)
+	if err != nil {
+		return err
+	}
+	m := &http.Server{Handler: metrics.Default, ReadHeaderTimeout: 5 * time.Second, ErrorLog: log0()}
+	go func() { _ = m.Serve(ln) }()
+	go func() {
+		<-ctx.Done()
+		_ = m.Close()
+	}()
+	return nil
+}
+
+// printMetrics prints the running service's /metrics.
+func printMetrics(args []string) error {
+	fs := flag.NewFlagSet("metrics", flag.ExitOnError)
+	addr := fs.String("addr", defaultMetricsAddr, "where the service answers /metrics")
+	_ = fs.Parse(args)
+	c := &http.Client{Timeout: 5 * time.Second}
+	resp, err := c.Get("http://" + *addr + "/metrics")
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	_, err = io.Copy(os.Stdout, resp.Body)
+	return err
+}
+
 func log0() *log.Logger { return log.New(io.Discard, "", 0) }
 
 // loadAPNs reads Apple's key from dir: key.p8, key-id and team-id, as the

cmd/ftwcloudd/main_test.go

@@ -0,0 +1,23 @@
+package main
+
+import (
+	"testing"
+	"time"
+)
+
+func TestTheRestartSpreadFollowsTheSockets(t *testing.T) {
+	for _, c := range []struct {
+		sockets, rate int
+		want          time.Duration
+	}{
+		{0, 500, 15 * time.Second},
+		{4000, 500, 15 * time.Second},
+		{12000, 500, 24 * time.Second},
+		{1_000_000, 500, 60 * time.Second},
+		{12000, 0, 24 * time.Second},
+	} {
+		if got := restartSpread(c.sockets, c.rate, 15*time.Second); got != c.want {
+			t.Errorf("restartSpread(%d, %d) = %s, want %s", c.sockets, c.rate, got, c.want)
+		}
+	}
+}