diff --git a/changes/unreleased/Added-20260809-120000.yaml b/changes/unreleased/Added-20260809-120000.yaml new file mode 100644 index 0000000..ba9c112 --- /dev/null +++ b/changes/unreleased/Added-20260809-120000.yaml @@ -0,0 +1,5 @@ +kind: Added +body: A data directory now carries a format version, and kvs refuses to start on one it does not recognise instead of replaying bytes another version laid out; the refusal names both versions and what to do about it, and it covers the Raft store as well as the append log +time: 2026-08-09T12:00:00.000000+09:00 +custom: + Issue: "245" diff --git a/content/docs/clustering.md b/content/docs/clustering.md index d3521bc..1f9f439 100644 --- a/content/docs/clustering.md +++ b/content/docs/clustering.md @@ -121,6 +121,14 @@ staying up, not for throughput. single-node append log rather than adding to it — the same changes written twice would only be two things to keep in step. +**The data directory carries a format version.** kvs writes a `format` file into `--data-dir` +the first time it uses one, and refuses to start on a directory whose version it does not +recognize — including one written before that file existed. The alternative is a replay reading +bytes laid out by another version, which surfaces much later and looks like corruption rather +than a version mismatch. There is no conversion: move the directory aside and load the data +again. The version covers the Raft store too, so it moves when the consensus library changes +its own layout. + **`--node-id` is required when there is no RESP listener.** It defaults to `--resp-addr`, so `--resp-addr none` leaves nothing to borrow and `serve` refuses to start rather than join a cluster under a blank name. diff --git a/content/docs/compatibility.md b/content/docs/compatibility.md index dba7cf9..903e101 100644 --- a/content/docs/compatibility.md +++ b/content/docs/compatibility.md @@ -74,10 +74,12 @@ package is not. **Anything under `internal/`.** The Go toolchain already stops you importing it; this is the same statement in words. -**The on-disk format.** The append log in `--data-dir`, and the Raft directory beside it, carry -no format version yet. Data written by one release is not promised to be readable by another. -Until that changes, treat an upgrade as: drain, start the new version against an empty -directory, reload. +**The on-disk format.** A data directory carries a `format` file naming the version that wrote +it, and kvs refuses to start on a directory whose version it does not recognize rather than +reading it and hoping. That refusal is the promise; the contents are not. Data written by one +release is not promised to be readable by another, and kvs does not convert between formats. An +upgrade that raises the format means: drain, start the new version against an empty directory, +load the data again. **Performance.** Throughput, latency, and memory are not part of the promise. kvs is built for not losing writes, not for being fast, and a release may trade one for the other. diff --git a/internal/cluster/cluster.go b/internal/cluster/cluster.go index 087221e..95ee35d 100644 --- a/internal/cluster/cluster.go +++ b/internal/cluster/cluster.go @@ -22,6 +22,7 @@ import ( raftboltdb "github.com/hashicorp/raft-boltdb/v2" "github.com/skyoo2003/kvs" + "github.com/skyoo2003/kvs/internal/datadir" ) const ( @@ -87,7 +88,13 @@ type Node struct { // Start brings this node up and, if asked, starts a new cluster around it. func Start(cfg Config, store *kvs.Store) (*Node, error) { - dir := filepath.Join(cfg.DataDir, "raft") + // A clustered node never goes through kvs.Open, so the format check has to happen here too + // or the Raft store would be handed a directory nobody vouched for. + if err := datadir.Ensure(cfg.DataDir); err != nil { + return nil, err + } + + dir := filepath.Join(cfg.DataDir, datadir.RaftName) if err := os.MkdirAll(dir, 0o750); err != nil { return nil, fmt.Errorf("create raft dir: %w", err) } diff --git a/internal/datadir/datadir.go b/internal/datadir/datadir.go new file mode 100644 index 0000000..cbbb217 --- /dev/null +++ b/internal/datadir/datadir.go @@ -0,0 +1,223 @@ +// Package datadir owns the shape of the directory kvs keeps its data in, and the version +// stamped on it. Every path that opens one checks that version before reading anything, so a +// directory this build cannot understand fails at startup with a reason rather than part-way +// through a replay, or worse, quietly. +package datadir + +import ( + "errors" + "fmt" + "io" + "os" + "path/filepath" + "strconv" + "strings" +) + +// Version is what this build writes and the only one it reads. Raise it whenever the bytes in +// a data directory change shape — including when a dependency that owns part of the directory, +// the Raft log store among them, changes its own format under us. There is no conversion +// between versions, which is why the check has to be loud. +const Version = 1 + +const ( + // FormatName is the file holding Version. It sits beside the data rather than inside it + // because the Raft store's files belong to a library that will not carry our header. + FormatName = "format" + + // LogName is the single-node append log, and RaftName the directory the Raft store keeps + // its own files in. Together they are how an unversioned directory is told from an empty + // one, so they live here rather than being spelled out wherever each is opened. + LogName = "kvs.log" + RaftName = "raft" +) + +// ErrFormat is what every refusal to open a data directory matches. Use errors.As with +// FormatError to find out what was actually on disk. +var ErrFormat = errors.New("data directory format") + +// FormatError says what the directory holds and what this build understands. Missing marks +// the case where there is no format file at all, which is not the same as one holding +// nothing, and Raw is whatever it did hold. The version is not kept as a number: zero is a +// perfectly good version to find written down, so it cannot double as "there was none". +type FormatError struct { + Dir string + Missing bool + Raw string +} + +func (e *FormatError) Error() string { + if e.Missing { + return fmt.Sprintf("%s holds kvs data but no %s file, so it was written before kvs "+ + "versioned its data directory; this build writes format %d. Move the directory "+ + "aside and start again.", e.Dir, FormatName, Version) + } + + if _, err := strconv.Atoi(e.Raw); err != nil { + return fmt.Sprintf("%s has a %s file holding %q, which is not a version number; this "+ + "build writes format %d. Move the directory aside and start again.", + e.Dir, FormatName, e.Raw, Version) + } + + return fmt.Sprintf("%s is format %s and this build understands format %d. kvs does not "+ + "convert between them: run the version that wrote it, or move the directory aside "+ + "and load the data again.", e.Dir, e.Raw, Version) +} + +func (e *FormatError) Is(target error) bool { + return target == ErrFormat +} + +// Ensure makes dir usable by this build, creating it and stamping the current version when it +// is new. A directory belonging to another version comes back as a FormatError and nothing is +// touched: refusing to start is the whole point, because the alternative is a replay that +// half-works. +func Ensure(dir string) error { + if err := os.MkdirAll(dir, 0o750); err != nil { + return fmt.Errorf("create data dir: %w", err) + } + + path := filepath.Join(dir, FormatName) + + raw, err := readFormat(path) + + switch { + case errors.Is(err, os.ErrNotExist): + // An empty directory is a new one. A directory already holding data without a version + // was written before this check existed, and guessing which version it is would be + // exactly the silent failure this package exists to prevent. + found, dataErr := holdsData(dir) + if dataErr != nil { + return dataErr + } + + if found { + return &FormatError{Dir: dir, Missing: true} + } + + return stamp(path) + + case err != nil: + return fmt.Errorf("read data dir format: %w", err) + } + + if len(raw) > maxFormatBytes { + // A file longer than the cap holds something other than a version, whatever its first + // bytes look like, so trusting the prefix would accept a directory on the strength of the + // part that happened to fit. Only that prefix is quoted back: the rest was never read. + return &FormatError{Dir: dir, Raw: strings.TrimSpace(string(raw[:maxFormatBytes])) + "..."} + } + + text := strings.TrimSpace(string(raw)) + + if found, convErr := strconv.Atoi(text); convErr != nil || found != Version { + return &FormatError{Dir: dir, Raw: text} + } + + return nil +} + +// maxFormatBytes is more room than a version number will ever need and less than is worth +// reading into memory. Something has replaced the file if it is longer, and the promise is a +// refusal that says so, not an allocation the size of whatever was put there. +const maxFormatBytes = 64 + +// readFormat returns what the format file holds, or as much of it as could be a version plus one +// byte, which is how the caller tells a marker that ran on from one that fit. The caller also +// tells a missing file from an unreadable one, so the error is returned unwrapped. +func readFormat(path string) ([]byte, error) { + //nolint:gosec // The path is the data directory the operator named; that is the feature. + file, err := os.Open(path) + if err != nil { + return nil, err + } + defer func() { _ = file.Close() }() + + return io.ReadAll(io.LimitReader(file, maxFormatBytes+1)) +} + +// holdsData reports whether the directory already carries a keyspace. It looks for the two +// things kvs puts there by name rather than asking whether the directory is empty, because a +// freshly mounted volume is not empty — lost+found is there before kvs ever runs. +func holdsData(dir string) (bool, error) { + for _, name := range []string{LogName, RaftName} { + // Lstat rather than Stat: a symlink whose target is a volume that is not mounted yet is + // still an entry kvs put there, and following it would report the data as absent, stamp + // the directory, and accept an unversioned keyspace once the volume came back. + switch _, err := os.Lstat(filepath.Join(dir, name)); { + case err == nil: + return true, nil + + case !errors.Is(err, os.ErrNotExist): + // Anything other than "it is not there" leaves the question open, and answering it + // with "no data" would stamp this build's version onto a directory that may hold an + // older one's keyspace — the silent acceptance this package exists to prevent. + return false, fmt.Errorf("look for %s in %s: %w", name, dir, err) + } + } + + return false, nil +} + +// stamp writes the version through a temporary file and renames it into place, so an +// interrupted write leaves no format file rather than an unreadable one. The difference +// matters: no file on an empty directory is stamped again next time, whereas a truncated one +// is a version this build does not recognize and would refuse the directory forever. +func stamp(path string) error { + dir := filepath.Dir(path) + + tmp, err := os.CreateTemp(dir, FormatName+".*") + if err != nil { + return fmt.Errorf("write data dir format: %w", err) + } + defer func() { _ = os.Remove(tmp.Name()) }() + + if err := writeAndSync(tmp, strconv.Itoa(Version)+"\n"); err != nil { + return fmt.Errorf("write data dir format: %w", err) + } + + if err := os.Rename(tmp.Name(), path); err != nil { + return fmt.Errorf("write data dir format: %w", err) + } + + // The rename is a directory entry, and syncing the file did not make it durable. Without + // this a crash can leave data that was flushed afterwards next to no marker at all, which + // reads as a directory written before kvs versioned them and is refused from then on. + return SyncDir(dir) +} + +func writeAndSync(file *os.File, text string) error { + if _, err := file.WriteString(text); err != nil { + _ = file.Close() + + return err + } + + if err := file.Sync(); err != nil { + _ = file.Close() + + return err + } + + return file.Close() +} + +// SyncDir flushes the directory's own entries, which syncing a file inside it does not cover. +// Anything that renames a file into a data directory has to call it, or a crash can leave the +// directory naming neither the old file nor the new one. It lives here because the directory is +// this package's business, and two copies would be two places to fix. +func SyncDir(dir string) error { + //nolint:gosec // The path is the data directory the operator named; that is the feature. + handle, err := os.Open(dir) + if err != nil { + return fmt.Errorf("open data dir: %w", err) + } + + if err := handle.Sync(); err != nil { + _ = handle.Close() + + return fmt.Errorf("sync data dir: %w", err) + } + + return handle.Close() +} diff --git a/internal/datadir/datadir_test.go b/internal/datadir/datadir_test.go new file mode 100644 index 0000000..50657cb --- /dev/null +++ b/internal/datadir/datadir_test.go @@ -0,0 +1,285 @@ +package datadir_test + +import ( + "errors" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/skyoo2003/kvs/internal/datadir" +) + +func TestEnsureStampsANewDirectory(t *testing.T) { + dir := t.TempDir() + + if err := datadir.Ensure(dir); err != nil { + t.Fatalf("Ensure() error = %v", err) + } + + //nolint:gosec // The path is the temporary directory this test just made. + raw, err := os.ReadFile(filepath.Join(dir, datadir.FormatName)) + if err != nil { + t.Fatalf("read format: %v", err) + } + + if got, want := strings.TrimSpace(string(raw)), "1"; got != want { + t.Errorf("format = %q, want %q", got, want) + } + + // The stamp is written under a temporary name and renamed into place. None of those names + // may survive it, or the next reader finds a directory holding two answers. + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("read dir: %v", err) + } + + for _, entry := range entries { + if entry.Name() != datadir.FormatName { + t.Errorf("%s left behind after stamping", entry.Name()) + } + } + + // Opening it again has to be the ordinary case, not a second stamping. + if err := datadir.Ensure(dir); err != nil { + t.Errorf("Ensure() on its own directory error = %v", err) + } +} + +// A stat that fails for any reason other than absence leaves the question open, and answering +// it with "nothing here" would stamp this build's version onto another one's keyspace. +func TestEnsureRefusesWhenItCannotTellWhetherDataIsThere(t *testing.T) { + dir := t.TempDir() + + // A directory that cannot be searched is one whose contents cannot be stat'd. It stands in for + // the transient filesystem error this guards against, which a test cannot arrange directly. + if err := os.Chmod(dir, 0); err != nil { + t.Skipf("cannot drop directory permissions here: %v", err) + } + + if _, probeErr := os.Lstat(filepath.Join(dir, datadir.LogName)); errors.Is(probeErr, os.ErrNotExist) { + _ = searchable(dir) + t.Skip("this filesystem still answers for an unsearchable directory") + } + + err := datadir.Ensure(dir) + + // Restored before the assertions below, which have to be able to look inside again — and + // before t.TempDir cleans up, which cannot remove what it cannot enter. + if chmodErr := searchable(dir); chmodErr != nil { + t.Fatalf("restore permissions: %v", chmodErr) + } + + if err == nil { + t.Fatal("Ensure() error = nil, want a refusal") + } + + if errors.Is(err, datadir.ErrFormat) { + t.Errorf("Ensure() error = %v, want the failure to look, not a verdict on the format", err) + } + + if _, statErr := os.Stat(filepath.Join(dir, datadir.FormatName)); statErr == nil { + t.Error("the directory was stamped without knowing what it holds") + } +} + +// A format file far larger than a version is refused for what it says, not loaded for its size. +func TestEnsureBoundsWhatItReadsFromTheFormatFile(t *testing.T) { + dir := t.TempDir() + writeFile(t, filepath.Join(dir, datadir.FormatName), strings.Repeat("x", 1<<20)) + + err := datadir.Ensure(dir) + + var formatErr *datadir.FormatError + if !errors.As(err, &formatErr) { + t.Fatalf("Ensure() error = %v, want a FormatError", err) + } + + // The read cap plus the marker saying the file went on, and nothing like the file itself. + if quoted := len(formatErr.Raw); quoted > 64+len("...") { + t.Errorf("Raw is %d bytes long, so the whole file was read into memory", quoted) + } +} + +// Data behind a symlink whose target is a volume that is not mounted yet is data all the same. +// Stamping the directory would accept it as this format the moment the volume came back. +func TestEnsureRefusesADanglingLegacyDataLink(t *testing.T) { + dir := t.TempDir() + + link := filepath.Join(dir, datadir.LogName) + if err := os.Symlink(filepath.Join(dir, "volume", datadir.LogName), link); err != nil { + t.Skipf("symlinks unavailable here: %v", err) + } + + err := datadir.Ensure(dir) + + var formatErr *datadir.FormatError + if !errors.As(err, &formatErr) { + t.Fatalf("Ensure() error = %v, want a FormatError", err) + } + + if !formatErr.Missing { + t.Errorf("Missing = false, want the refusal for data written before versioning") + } + + if _, statErr := os.Stat(filepath.Join(dir, datadir.FormatName)); statErr == nil { + t.Error("the directory was stamped over data whose volume is not mounted") + } +} + +func TestEnsureCreatesADirectoryThatIsNotThereYet(t *testing.T) { + dir := filepath.Join(t.TempDir(), "nested", "kvs") + + if err := datadir.Ensure(dir); err != nil { + t.Fatalf("Ensure() error = %v", err) + } + + if _, err := os.Stat(filepath.Join(dir, datadir.FormatName)); err != nil { + t.Errorf("format file after Ensure: %v", err) + } +} + +// A directory holding data kvs cannot place is the case the whole package exists for. Each of +// these has to come back as ErrFormat, and the message has to say enough to act on. +func TestEnsureRefusesADirectoryItCannotRead(t *testing.T) { + tests := []struct { + name string + arrange func(t *testing.T, dir string) + // wants are substrings the message must carry, so that a reader is told what is on + // disk and what this build wanted. + wants []string + }{ + { + name: "a later format", + arrange: func(t *testing.T, dir string) { + t.Helper() + writeFormat(t, dir, "2") + }, + wants: []string{"format 2", "format 1"}, + }, + { + name: "an earlier format", + arrange: func(t *testing.T, dir string) { + t.Helper() + writeFormat(t, dir, "0") + }, + wants: []string{"format 0", "format 1"}, + }, + { + name: "something that is not a version", + arrange: func(t *testing.T, dir string) { + t.Helper() + writeFormat(t, dir, "banana") + }, + wants: []string{`"banana"`, "not a version number"}, + }, + { + // A file that exists and says nothing is not the same as no file, and the advice + // for the two differs. + name: "a format file holding nothing", + arrange: func(t *testing.T, dir string) { + t.Helper() + writeFile(t, filepath.Join(dir, datadir.FormatName), "") + }, + wants: []string{`holding ""`, "not a version number"}, + }, + { + // A marker whose first bytes read as this version but which goes on past what a + // version can be. Taking the prefix for the whole would accept a replaced file. + name: "a marker that runs past the read cap", + arrange: func(t *testing.T, dir string) { + t.Helper() + writeFile(t, filepath.Join(dir, datadir.FormatName), "1"+strings.Repeat(" ", 63)+"7") + }, + wants: []string{`"1..."`, "not a version number"}, + }, + { + name: "an append log written before versioning", + arrange: func(t *testing.T, dir string) { + t.Helper() + writeFile(t, filepath.Join(dir, datadir.LogName), "") + }, + wants: []string{"no format file", "format 1"}, + }, + { + name: "a raft directory written before versioning", + arrange: func(t *testing.T, dir string) { + t.Helper() + + if err := os.MkdirAll(filepath.Join(dir, datadir.RaftName), 0o750); err != nil { + t.Fatalf("create raft dir: %v", err) + } + }, + wants: []string{"no format file", "format 1"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + dir := t.TempDir() + tt.arrange(t, dir) + + err := datadir.Ensure(dir) + if err == nil { + t.Fatal("Ensure() error = nil, want a refusal") + } + + if !errors.Is(err, datadir.ErrFormat) { + t.Errorf("errors.Is(err, ErrFormat) = false, err = %v", err) + } + + var formatErr *datadir.FormatError + if !errors.As(err, &formatErr) { + t.Fatalf("errors.As(err, *FormatError) = false, err = %v", err) + } + + if formatErr.Dir != dir { + t.Errorf("Dir = %q, want %q", formatErr.Dir, dir) + } + + for _, want := range tt.wants { + if !strings.Contains(err.Error(), want) { + t.Errorf("message %q does not carry %q", err.Error(), want) + } + } + }) + } +} + +// A volume kvs has never touched is rarely empty, and refusing to start on one would be a +// worse failure than the one this package prevents. +func TestEnsureIgnoresFilesThatAreNotOurs(t *testing.T) { + dir := t.TempDir() + + if err := os.MkdirAll(filepath.Join(dir, "lost+found"), 0o750); err != nil { + t.Fatalf("create lost+found: %v", err) + } + + writeFile(t, filepath.Join(dir, "README"), "not ours") + + if err := datadir.Ensure(dir); err != nil { + t.Fatalf("Ensure() error = %v", err) + } +} + +func writeFormat(t *testing.T, dir, text string) { + t.Helper() + + writeFile(t, filepath.Join(dir, datadir.FormatName), text+"\n") +} + +// searchable puts back the mode t.TempDir made the directory with, so a test that took it away +// can look inside again and the cleanup can still remove it. +// +//nolint:gosec // A directory has to carry its execute bit or nothing can enter it, this test included. +func searchable(dir string) error { + return os.Chmod(dir, 0o700) +} + +func writeFile(t *testing.T, path, text string) { + t.Helper() + + if err := os.WriteFile(path, []byte(text), 0o600); err != nil { + t.Fatalf("write %s: %v", path, err) + } +} diff --git a/kvs.go b/kvs.go index 794e937..dc83f1c 100644 --- a/kvs.go +++ b/kvs.go @@ -14,6 +14,8 @@ import ( "sync" "sync/atomic" "time" + + "github.com/skyoo2003/kvs/internal/datadir" ) var ( @@ -86,6 +88,13 @@ func Open(dir string, codec Codec) (*Store, error) { codec = StringCodec{} } + // Before anything is read: a directory written by a version whose format this build does + // not know has to stop here, where the reason can still be explained, rather than during + // the replay where it would look like corruption. + if err := datadir.Ensure(dir); err != nil { + return nil, err + } + lg, err := openLog(dir) if err != nil { return nil, err diff --git a/log.go b/log.go index a9ca044..b468d89 100644 --- a/log.go +++ b/log.go @@ -10,6 +10,8 @@ import ( "os" "path/filepath" "time" + + "github.com/skyoo2003/kvs/internal/datadir" ) // ErrUnsupportedValue is what a Codec returns for a value it cannot encode. Failing the write @@ -57,8 +59,10 @@ const ( opFlush = "flush" ) -// logName is the file the append log lives in, inside the directory Open was given. -const logName = "kvs.log" +// logName is the file the append log lives in, inside the directory Open was given. The name +// belongs to datadir because that package has to recognize a directory holding one, and two +// spellings of it would mean a log this build writes and the next one fails to notice. +const logName = datadir.LogName // record is one durable change. Value holds the encoded form and value the original: encoding // is deferred to commit so that a value the codec cannot handle fails the write that stored @@ -192,7 +196,7 @@ func (l *appendLog) rewrite(lines frame) error { } // The rename itself has to reach the disk, or a crash can leave the directory naming // neither the old file nor the new one. - if syncErr := syncDir(dir); syncErr != nil { + if syncErr := datadir.SyncDir(dir); syncErr != nil { return syncErr } @@ -240,18 +244,3 @@ func writeLines(file *os.File, lines frame) error { return nil } - -func syncDir(dir string) error { - //nolint:gosec // The path is the data directory the operator named; that is the feature. - handle, err := os.Open(dir) - if err != nil { - return fmt.Errorf("open data dir: %w", err) - } - defer func() { _ = handle.Close() }() - - if err := handle.Sync(); err != nil { - return fmt.Errorf("flush data dir: %w", err) - } - - return nil -} diff --git a/log_test.go b/log_test.go index 9a41764..6924c32 100644 --- a/log_test.go +++ b/log_test.go @@ -5,8 +5,11 @@ import ( "errors" "os" "path/filepath" + "strings" "testing" "time" + + "github.com/skyoo2003/kvs/internal/datadir" ) func TestStringCodecRoundTrip(t *testing.T) { @@ -202,6 +205,44 @@ func TestOpenDefaultsToStringCodec(t *testing.T) { } // A store with no log is still the default, and must not have gained a file or a failure. +// A data directory another version wrote has to stop Open before the replay, and say so. The +// replay reading someone else's bytes is the failure this check exists to prevent, and it +// would arrive looking like corruption. +func TestOpenRefusesADirectoryFromAnotherFormat(t *testing.T) { + dir := t.TempDir() + + store, err := Open(dir, nil) + if err != nil { + t.Fatalf("Open() error = %v", err) + } + if err := store.Put("language", "go"); err != nil { + t.Fatalf("Put() error = %v", err) + } + if err := store.Close(); err != nil { + t.Fatalf("Close() error = %v", err) + } + + format := filepath.Join(dir, datadir.FormatName) + if err := os.WriteFile(format, []byte("2\n"), 0o600); err != nil { + t.Fatalf("write format: %v", err) + } + + _, reopenErr := Open(dir, nil) + if reopenErr == nil { + t.Fatal("Open() error = nil, want a refusal") + } + + if !errors.Is(reopenErr, datadir.ErrFormat) { + t.Errorf("errors.Is(err, datadir.ErrFormat) = false, err = %v", reopenErr) + } + + for _, want := range []string{"format 2", "format 1"} { + if !strings.Contains(reopenErr.Error(), want) { + t.Errorf("message %q does not carry %q", reopenErr.Error(), want) + } + } +} + func TestNewStoreWritesNothing(t *testing.T) { dir := t.TempDir()