diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index e153378..b28bc88 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -44,6 +44,13 @@ make build # build binary to dist/ make clean # remove dist/ ``` +`make all` is what CI runs. The long one is separate and opt-in: + +```sh +make soak # 5 minutes of load with a node restarted every 30 seconds +make soak SOAK=4h # the full run the numbers in the docs come from +``` + ### Pre-commit Hooks (optional) ```sh diff --git a/Makefile b/Makefile index 43c7621..0ee48de 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,7 @@ -.PHONY: all setup clean build test lint lint-fix vet coverage race +.PHONY: all setup clean build test lint lint-fix vet coverage race soak + +# How long `make soak` runs for. The full run behind the numbers in the docs is SOAK=4h. +SOAK ?= 5m all: vet lint test build @@ -30,3 +33,11 @@ coverage: race: @go test -race ./... + +# Load and fault injection over hours. Only the two packages naming -soak are listed, because +# passing an unknown flag to the others fails them before they start, and they have to come +# before -soak: go test treats everything after a flag it does not know as arguments for the +# test binary and falls back to the current directory. -timeout 0 leaves the deadline to the +# test, so raising SOAK never means recalculating a second number. +soak: + @go test . ./internal/cluster -run TestSoak -soak $(SOAK) -timeout 0 -v diff --git a/changes/unreleased/Documentation-20260809-210000.yaml b/changes/unreleased/Documentation-20260809-210000.yaml new file mode 100644 index 0000000..c615385 --- /dev/null +++ b/changes/unreleased/Documentation-20260809-210000.yaml @@ -0,0 +1,5 @@ +kind: Documentation +body: The durability and clustering page now carries measured numbers from a four hour run under load with a node stopped every thirty seconds and kept down for ten - 329,631 acknowledged writes, 111,516 of them taken while a node was gone, and none of them lost across 479 restarts - along with 51 bytes of append log per write, the Raft log a node too busy restarting never truncates, and a make soak target that reproduces all of it +time: 2026-08-09T21:00:00.000000+09:00 +custom: + Issue: "246" diff --git a/content/docs/clustering.md b/content/docs/clustering.md index 1f9f439..d7442ce 100644 --- a/content/docs/clustering.md +++ b/content/docs/clustering.md @@ -35,6 +35,15 @@ the file without them. keyspace in memory, so rewriting the log costs nothing extra at that moment and needs no background worker. A long-running process grows its log without bound in the meantime. +How fast is worth knowing before you size a disk. Four hours of continuous writes over a +thousand keys — 1,726,455 of them, so every key was overwritten some seventeen hundred times — +left an **87MB log for a keyspace that never exceeded a thousand entries**, at **51 bytes per +write**. That last figure is this workload's, not yours: a record carries the key and the +encoded value, so your own average record is what scales it. The shape is what carries over — +the log grows with writes, not with time and not with how much data you are keeping. Memory +settled over the same four hours, 1.11MB to 1.22MB, so it is the disk that needs watching, and +a restart is what reclaims it. + **One node means one disk.** Durability is not availability: a lost disk is lost data, and a stopped process is a stopped service. That is what clustering is for. @@ -69,6 +78,19 @@ eventual: the price is that every write pays one consensus round. **Failover needs no person.** Stop the leader and the survivors elect a replacement on their own. +**Nothing acknowledged was lost over four hours of being knocked down.** A three-node cluster +took 329,631 writes with one node stopped every thirty seconds and left down for ten — 479 +restarts — and **111,516 of those writes were taken while a node was gone**. Each node that came +back was held against the latest acknowledged value of every key before the load was allowed to +write again, and all three were checked once more at the end. Nothing was missing and nothing +crashed. The check is per key rather than per write: an interval that gets through more writes +than there are keys overwrites some of its own, and only the later value is left to compare, so +the run reports how many acknowledged writes were superseded before a check could see them. +Writes were +refused slightly more often than they were taken, which is what asking a follower looks like — +the refusal names the leader — rather than anything being lost: only a write that returns is a +write kvs claims to have. `make soak SOAK=4h` is that run. + ### What it does not **Three nodes, not two.** A majority of two is two, so a two-node cluster stops accepting @@ -115,6 +137,25 @@ liveness checks; treat them as "this process is up", not "this node is useful". hold. Writes are serialized through the leader, one consensus round each: this is built for staying up, not for throughput. +**A node that restarts faster than it can snapshot never truncates its log.** Raft discards log +entries only once a snapshot covers them. It considers taking one every two to four minutes, and +takes it only if 8,192 entries have arrived since the last — at the write rate above that is +another five or six minutes — then keeps the most recent 10,240 entries anyway. In the run above +each node was going down every ninety seconds and so never lived long enough to take one: +after four hours every node held **126MB of Raft log for a thousand keys**, none of it +discardable. A node in a crash loop fills its disk while looking like it is merely restarting. +A node that stays up snapshots normally and none of this arises. + +**Memory holds under that churn, unless the node comes back instantly.** Across those four +hours the heap of the one process holding all three nodes ended where it began, 4.3MB once +settled and 4.0MB after 479 restarts. An earlier version of the same run restarted each node +the moment it stopped rather than leaving it down, and that one climbed from 10MB to 139MB over +four hours; a heap profile put more than half of what was still reachable in Raft's own network +transport, which holds a 256KB read buffer and a 256KB write buffer per connection, while the +only frame belonging to kvs held 544KB — the thousand keys themselves. Ten seconds down was +enough for none of that to accumulate. It is the process that crash-loops without pause that +grows, and it wants a memory limit as much as it wants the disk headroom above. + ## Configuration notes **`--data-dir` is required in a cluster.** The Raft log has to live somewhere. It replaces the diff --git a/content/docs/contributing.md b/content/docs/contributing.md index 301b7c5..e508e11 100644 --- a/content/docs/contributing.md +++ b/content/docs/contributing.md @@ -21,6 +21,14 @@ make lint # run golangci-lint make build # build binary ``` +`make all` is what CI runs. The soak test is separate and opt-in, because it is measured in +hours rather than seconds: + +```sh +make soak # 5 minutes of load with a node restarted every 30 seconds +make soak SOAK=4h # the full run the numbers on the clustering page come from +``` + ## PR Guidelines - Keep PRs small and focused on a single concern diff --git a/internal/cluster/soak_test.go b/internal/cluster/soak_test.go new file mode 100644 index 0000000..f409a2b --- /dev/null +++ b/internal/cluster/soak_test.go @@ -0,0 +1,305 @@ +package cluster + +import ( + "errors" + "flag" + "io/fs" + "path/filepath" + "runtime" + "strconv" + "testing" + "time" + + "github.com/skyoo2003/kvs" + "github.com/skyoo2003/kvs/internal/datadir" +) + +// soakFor turns the long run on and says how long it lasts. Zero, the default, skips it, so +// `go test ./...` costs what it costs today and only `make soak` pays for this. +var soakFor = flag.Duration("soak", 0, "run the soak test for this long") + +// soakKeys is a fixed set that gets overwritten rather than one that grows. A keyspace that +// grows makes the heap grow for honest reasons, and then the heap says nothing about leaks. +const soakKeys = 1000 + +// killEvery is how often a node is taken away. One at a time: three nodes still hold a +// majority with one gone, and taking two would measure a design decision, not a fault. +const killEvery = 30 * time.Second + +// downFor is how long it stays away, and writes carry on the whole time. Restarting it straight +// away would leave it nothing to catch up on, and catching up is the part a restart has to +// survive; a run that never writes into the gap proves only that a process can be started twice. +const downFor = 10 * time.Second + +// Hours of writes with a node taken away every half minute and kept away while the writes keep +// coming, held after every restart and again at the end to every value kvs acknowledged. What +// it reports - writes, restarts, heap, bytes on disk - is the record that says the thing +// survived, so those numbers go in the docs. +func TestSoakSurvivesLoadAndFailover(t *testing.T) { + if *soakFor == 0 { + t.Skip("soak run is off; pass -soak , or run make soak") + } + + nodes := startCluster(t, 3) + waitForLeader(t, nodes) + + run := &soakRun{nodes: nodes, acked: make(map[string]string, soakKeys)} + run.load(t, *soakFor) + run.verify(t) + run.report(t) +} + +type soakRun struct { + nodes []*testNode + // acked holds the last value each key was acknowledged with. A write that came back with + // an error is a retry, not a loss, so only what returned nil is in here. + acked map[string]string + // last is the key of the most recent acknowledged write, used as the barrier to wait on. + last string + // down is the node currently stopped, if any, and backAt when it is due to come back. + down *testNode + backAt time.Time + writes int + // shorthanded counts the acknowledged writes taken while a node was stopped. It is the + // claim the docs make, so it is counted rather than inferred from the timings. + shorthanded int + refused int + failovers int + // window counts the writes acknowledged since the last check, and unverified the ones a check + // could not vouch for on its own. The load walks the keys in order, so a window longer than + // the keyspace overwrites values it already acknowledged and only the newer one is left to + // compare. Zero unverified means every acknowledged write was held against a returning node + // as itself, rather than as whatever replaced it. + window int + unverified int + // warmup and final are the heap once the run has settled and at the end of it. + warmup heapSample + final heapSample +} + +// heapSample is what the run records at its endpoints. alloc is what is still reachable, which +// is the question a leak asks. inUse is what the spans hold, which is closer to what an +// operator sees and which a non-moving collector does not hand back just because objects died. +type heapSample struct { + alloc uint64 + inUse uint64 +} + +func (r *soakRun) load(t *testing.T, over time.Duration) { + t.Helper() + + start := time.Now() + deadline := start.Add(over) + // The heap is first read once the run has settled, so the baseline is a steady state + // rather than a process still filling its caches. + warmupAt := start.Add(over / 10) + nextKill := start.Add(killEvery) + + for round := 0; time.Now().Before(deadline); round++ { + r.put(t, "soak-"+strconv.Itoa(round%soakKeys), strconv.Itoa(round)) + + now := time.Now() + // A live process never has a zero heap, so zero is still "not read yet". + if r.warmup.alloc == 0 && now.After(warmupAt) { + r.warmup = heapAfterGC() + } + + switch { + case r.down != nil && now.After(r.backAt): + r.bringBack(t) + + case r.down == nil && now.After(nextKill): + r.takeDown(t, r.failovers%len(r.nodes)) + nextKill = now.Add(killEvery) + } + } + + if r.down != nil { + r.bringBack(t) + } + + r.final = heapAfterGC() +} + +// put offers the write to every node until one takes it, which is what a client following a +// MOVED does. Only a write that returned nil is recorded, because that is the only one kvs +// claimed to have. +func (r *soakRun) put(t *testing.T, key, value string) { + t.Helper() + + eventually(t, "a node to take "+key, func() bool { + for _, node := range r.nodes { + // A node known to be stopped is skipped rather than asked, so that a refusal means + // what it says: the leader was somewhere else. + if node == r.down { + continue + } + + err := node.store.Put(key, value) + + // A refusal names the leader and is the ordinary answer from a follower, so the next + // node gets asked. Anything else - an apply that timed out, a store that failed, an + // FSM that would not take the frame - is not a redirect, and retrying until some node + // says yes would let a run that is quietly broken finish looking clean. + if errors.Is(err, kvs.ErrNotLeader) { + r.refused++ + + continue + } + + if err != nil { + t.Fatalf("%s Put(%q) = %v, want nil or %v", node.id, key, err, kvs.ErrNotLeader) + } + + r.acked[key] = value + r.last = key + r.writes++ + r.window++ + + if r.down != nil { + r.shorthanded++ + } + + return true + } + + return false + }) +} + +// takeDown stops a node and leaves it stopped. The caller keeps writing, so what the cluster is +// asked to do is take writes a node short and hand them over when it comes back. +func (r *soakRun) takeDown(t *testing.T, i int) { + t.Helper() + + r.down = r.nodes[i] + r.down.stop(t) + r.backAt = time.Now().Add(downFor) +} + +// bringBack restarts the stopped node and checks it against every acknowledged value before the +// run is allowed to write again. Checking here rather than only at the end is what keeps the +// evidence: a value lost while the node was away would otherwise be overwritten by a later +// round, and the missing write would be gone from the record as well as from disk. It narrows +// that window to one interval rather than closing it, which is what unverified counts. +func (r *soakRun) bringBack(t *testing.T) { + t.Helper() + + node := r.down + node.restart(t) + r.down = nil + r.failovers++ + + // A restart starts the node on an empty store, so this waits out a replay of everything, + // not a top-up. Raft applies its log in order, so holding the last write means holding the + // rest, and the comparison that follows is a comparison rather than a race. + node.mustEventuallyHold(t, r.last, r.acked[r.last]) + r.mustHoldEverything(t, node) + r.closeWindow() + + // Sampled every restart rather than only at the ends, because two readings cannot tell a + // leak from a step change. No forced collection here: this is a trend, and stopping the + // world hundreds of times would change what it is measuring. + var stats runtime.MemStats + runtime.ReadMemStats(&stats) + t.Logf("restart %d of %s: heap alloc %d bytes, heap in use %d bytes, %d goroutines", + r.failovers, node.id, stats.HeapAlloc, stats.HeapInuse, runtime.NumGoroutine()) +} + +// verify is what the run exists for: every value kvs acknowledged has to be on every node when +// the load stops. It is the last of the checks rather than the only one - each restart has +// already been held to the same standard against the state at that moment. +func (r *soakRun) verify(t *testing.T) { + t.Helper() + + for _, node := range r.nodes { + node.mustEventuallyHold(t, r.last, r.acked[r.last]) + r.mustHoldEverything(t, node) + } + + r.closeWindow() +} + +// closeWindow settles what the check just made could account for. The load writes the keys in +// order, so a window that ran past the keyspace acknowledged some keys twice, and the earlier +// value of each is no longer anywhere to compare - not a loss, but not evidence either. +func (r *soakRun) closeWindow() { + if r.window > soakKeys { + r.unverified += r.window - soakKeys + } + + r.window = 0 +} + +func (r *soakRun) mustHoldEverything(t *testing.T, node *testNode) { + t.Helper() + + for key, want := range r.acked { + got, err := node.store.Get(key) + if err != nil || got != want { + t.Errorf("%s Get(%q) = %v, %v, want %v, nil", node.id, key, got, err, want) + } + } +} + +func (r *soakRun) report(t *testing.T) { + t.Helper() + + t.Logf("ran %s: %d writes acknowledged over %d keys, %d refused, %d node restarts", + *soakFor, r.writes, len(r.acked), r.refused, r.failovers) + t.Logf("%d of those writes were taken while a node was stopped, each stopped for %s", + r.shorthanded, downFor) + t.Logf("%d acknowledged writes were superseded before a check could compare them", r.unverified) + t.Logf("heap alloc: %d bytes once settled, %d bytes at the end", r.warmup.alloc, r.final.alloc) + t.Logf("heap in use: %d bytes once settled, %d bytes at the end, %d goroutines left", + r.warmup.inUse, r.final.inUse, runtime.NumGoroutine()) + + for _, node := range r.nodes { + dir := filepath.Join(node.dir, datadir.RaftName) + t.Logf("%s raft store: %d bytes", node.id, dirBytes(t, dir)) + } +} + +// The heap is reported and not asserted on, which is a decision rather than an omission. It +// held over four hours and 479 restarts, 4.3MB settled to 4.0MB at the end. An earlier version +// of this test brought each node back the instant it stopped, and that one went from 10MB to +// 139MB over the same four hours, with more than half of what was still reachable sitting in +// Raft's own NetworkTransport - a 256KB read buffer and a 256KB write buffer per connection. +// The only frame belonging to kvs held 544KB, which is the thousand keys themselves. A +// threshold tuned to pass one of those runs says nothing about the other, so what stays is the +// measurement, and clustering.md carries what it means. The assertion that matters is the +// comparison at every restart: no acknowledged value went missing across 479 of them. + +func heapAfterGC() heapSample { + // Collect first: the question is what is retained, not what has yet to be swept. + runtime.GC() + + var stats runtime.MemStats + runtime.ReadMemStats(&stats) + + return heapSample{alloc: stats.HeapAlloc, inUse: stats.HeapInuse} +} + +func dirBytes(t *testing.T, dir string) int64 { + t.Helper() + + var total int64 + err := filepath.WalkDir(dir, func(_ string, entry fs.DirEntry, err error) error { + if err != nil || entry.IsDir() { + return err + } + + info, err := entry.Info() + if err != nil { + return err + } + total += info.Size() + + return nil + }) + if err != nil { + t.Fatalf("walk %s: %v", dir, err) + } + + return total +} diff --git a/soak_test.go b/soak_test.go new file mode 100644 index 0000000..b710537 --- /dev/null +++ b/soak_test.go @@ -0,0 +1,128 @@ +package kvs + +import ( + "flag" + "os" + "path/filepath" + "runtime" + "strconv" + "testing" + "time" +) + +// soakFor turns the long run on and says how long it lasts. Zero, the default, skips it. +var soakFor = flag.Duration("soak", 0, "run the soak test for this long") + +// soakKeys is a fixed set that gets overwritten, so the keyspace stays the same size while the +// log does not. That difference is the measurement. +const soakKeys = 1000 + +// soakSizeCap stops the run before it can fill the disk it is measuring. Reaching it is a +// result rather than a failure: it says how long a process writing this fast has before the +// log is that large. It assumes the temporary directory has that much to spare - a smaller +// filesystem fails the write instead, which says so plainly enough. +const soakSizeCap = 2 << 30 + +// soakSizeEvery is how many writes pass between size checks. Statting the log on every write +// would measure the stat. +const soakSizeEvery = 10000 + +// The append log is compacted when Open replays it and nowhere else, so a process that stays +// up keeps appending - clustering.md says so, and this puts a number on it. A rate is more use +// to whoever has to size a disk than the sentence alone. Changing that behavior is not v1 +// work; measuring it is. +func TestSoakMeasuresLogGrowth(t *testing.T) { + if *soakFor == 0 { + t.Skip("soak run is off; pass -soak , or run make soak") + } + + dir := t.TempDir() + store := openTestStore(t, dir) + + start := time.Now() + deadline := start.Add(*soakFor) + // The heap is first read once the run has settled, so the baseline is a steady state + // rather than a process still filling its caches. + warmupAt := start.Add(*soakFor / 10) + + var writes int + var warmup uint64 + var capped bool + + for ; time.Now().Before(deadline); writes++ { + key := "soak-" + strconv.Itoa(writes%soakKeys) + if err := store.Put(key, strconv.Itoa(writes)); err != nil { + t.Fatalf("Put(%q) error = %v", key, err) + } + + now := time.Now() + // A live process never has a zero heap, so zero is still "not read yet". + if warmup == 0 && now.After(warmupAt) { + warmup = heapInUse() + } + + if writes%soakSizeEvery == 0 && logBytes(t, dir) > soakSizeCap { + capped = true + + break + } + } + + reportGrowth(t, dir, writes, time.Since(start), capped) + assertHeapSettled(t, warmup, heapInUse()) +} + +func reportGrowth(t *testing.T, dir string, writes int, elapsed time.Duration, capped bool) { + t.Helper() + + size := logBytes(t, dir) + if capped { + t.Logf("stopped early at the %d byte cap, %s into a %s run", soakSizeCap, elapsed, *soakFor) + } + + t.Logf("ran %s: %d writes over %d keys, log is %d bytes (%.1f bytes per write)", + elapsed.Round(time.Second), writes, soakKeys, size, float64(size)/float64(writes)) + t.Logf("log growth: %.1f MiB/hour at this write rate", float64(size)/(1<<20)/elapsed.Hours()) +} + +// assertHeapSettled is the only assertion here. The log is expected to grow; the keyspace is +// not, so what holds it has to settle. Half again is the loosest reading that still catches +// something holding on to every write. +func assertHeapSettled(t *testing.T, warmup, final uint64) { + t.Helper() + + // A run that hit the size cap before the baseline was due has nothing to compare, and a + // baseline of zero would turn the result the cap exists to produce into a failure. + if warmup == 0 { + t.Logf("heap in use: %d bytes at the end, with no settled reading to compare it to", final) + + return + } + + t.Logf("heap in use: %d bytes once settled, %d bytes at the end", warmup, final) + + if limit := warmup + warmup/2; final > limit { + t.Errorf("heap in use went from %d to %d bytes, want at most %d", warmup, final, limit) + } +} + +func heapInUse() uint64 { + // Collect first: the question is what is retained, not what has yet to be swept. + runtime.GC() + + var stats runtime.MemStats + runtime.ReadMemStats(&stats) + + return stats.HeapInuse +} + +func logBytes(t *testing.T, dir string) int64 { + t.Helper() + + info, err := os.Stat(filepath.Join(dir, logName)) + if err != nil { + t.Fatalf("stat log: %v", err) + } + + return info.Size() +}