Checkpoints: a copy of the store that opens as it was, at /_checkpoint

server/e2e/backup_test.go

@@ -0,0 +1,188 @@
+package e2e
+
+// Backups: a checkpoint taken through POST /_checkpoint while writes go on
+// opens, on a second server with the same key, as the store was: the
+// restore drill.
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+)
+
+const checkpointToken = "e2e-checkpoint-token"
+
+func checkpointPost(t *testing.T, s *server, token string) (int, string) {
+	t.Helper()
+	req, _ := http.NewRequest("POST", "http://"+s.monitorAddr+"/_checkpoint", nil)
+	if token != "" {
+		req.Header.Set("Authorization", "Bearer "+token)
+	}
+	resp, err := (&http.Client{Timeout: 60 * time.Second}).Do(req)
+	if err != nil {
+		t.Fatalf("checkpoint: %v", err)
+	}
+	defer resp.Body.Close()
+	b, _ := io.ReadAll(resp.Body)
+	return resp.StatusCode, strings.TrimSpace(string(b))
+}
+
+// storedOn returns the messages stored on topic, relayed from s: at least n
+// of them, and any more that come before the relay goes quiet. A topic can
+// hold more than n (writes taken during a checkpoint), and the relay
+// doesn't send them oldest first.
+func storedOn(t *testing.T, s *server, cid, topic string, n int) map[int]string {
+	t.Helper()
+	sub, err := dial(context.Background(), s.tcpAddr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer sub.close()
+	if _, err := sub.connect(cid, true, nextSess()); err != nil {
+		t.Fatal(err)
+	}
+	rid, err := sub.relay(topic, "24h")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !sub.waitAck(rid, 5*time.Second) {
+		t.Fatalf("no relay ack")
+	}
+	got := map[int]string{}
+	deadline := time.Now().Add(30 * time.Second)
+	for {
+		wait := time.Until(deadline)
+		if len(got) >= n {
+			wait = 500 * time.Millisecond
+		}
+		msg, ok := sub.waitPub(wait)
+		if !ok {
+			break
+		}
+		for _, m := range msg.Messages {
+			if seq, body, ok := decodePayload(m.Payload); ok {
+				got[seq] = string(body)
+			}
+		}
+	}
+	return got
+}
+
+func TestBackupCheckpointRestores(t *testing.T) {
+	dir := t.TempDir()
+	env := []string{"CHECKPOINT_DIR=" + dir, "CHECKPOINT_TOKEN=" + checkpointToken, "CHECKPOINT_MIN_MINUTES=0", "CHECKPOINT_KEEP=2"}
+	s := startServerWith(t, serverOpts{env: env})
+	ctx := context.Background()
+	cid := newClientID(0x0e2eb4c7)
+	topic := "groups.backup.x.message"
+
+	pub, err := dial(ctx, s.tcpAddr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer pub.close()
+	if _, err := pub.connect(cid, true, nextSess()); err != nil {
+		t.Fatal(err)
+	}
+	n := scaled(50)
+	for i := 0; i < n; i++ {
+		id, err := pub.publish(1, topic, encodePayload(i, fmt.Sprintf("m%d", i)), "24h")
+		if err != nil {
+			t.Fatal(err)
+		}
+		if !pub.waitAck(id, 5*time.Second) {
+			t.Fatalf("publish %d: no RECEIPT", i)
+		}
+	}
+	// A subscription made and dropped before the checkpoint: a delete.
+	if _, err := pub.subscribe(1, topic); err == nil {
+		pub.unsubscribe(topic)
+	}
+
+	// The token is needed, and writes go on while it runs.
+	if code, _ := checkpointPost(t, s, ""); code != http.StatusUnauthorized {
+		t.Errorf("no token: %d, want 401", code)
+	}
+	if code, _ := checkpointPost(t, s, "wrong"); code != http.StatusUnauthorized {
+		t.Errorf("a wrong token: %d, want 401", code)
+	}
+	var wg sync.WaitGroup
+	stop := make(chan struct{})
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for i := n; ; i++ {
+			select {
+			case <-stop:
+				return
+			default:
+			}
+			if _, err := pub.publish(1, topic, encodePayload(i, fmt.Sprintf("m%d", i)), "24h"); err != nil {
+				return
+			}
+			time.Sleep(20 * time.Millisecond)
+		}
+	}()
+	code, body := checkpointPost(t, s, checkpointToken)
+	close(stop)
+	wg.Wait()
+	if code != http.StatusOK {
+		t.Fatalf("checkpoint: %d %s\nlogs:\n%s", code, body, s.logs.String())
+	}
+	var ck struct {
+		Dir     string  `json:"dir"`
+		Seconds float64 `json:"seconds"`
+	}
+	if err := json.Unmarshal([]byte(body), &ck); err != nil || !strings.HasPrefix(ck.Dir, dir) {
+		t.Fatalf("checkpoint answer %q", body)
+	}
+	t.Logf("checkpoint in %.2fs at %s", ck.Seconds, ck.Dir)
+
+	// The metrics say so.
+	if _, metrics := monitorGet(t, s, "/_metrics"); !strings.Contains(metrics, "unitdb_checkpoint_last_success_timestamp_seconds ") ||
+		strings.Contains(metrics, "unitdb_checkpoint_last_success_timestamp_seconds 0\n") ||
+		!strings.Contains(metrics, "unitdb_store_messages ") {
+		t.Errorf("metrics after a checkpoint:\n%s", metrics)
+	}
+
+	// The drill: a second server on the checkpoint, with the same key.
+	restored := startServerWith(t, serverOpts{args: []string{"-db_path", ck.Dir}})
+	got := storedOn(t, restored, cid, topic, n)
+	for i := 0; i < n; i++ {
+		if want := fmt.Sprintf("m%d", i); got[i] != want {
+			t.Fatalf("restored seq %d: %q, want %q (got %d messages)\nlogs:\n%s", i, got[i], want, len(got), restored.logs.String())
+		}
+	}
+	t.Logf("restored %d messages, %d written during the checkpoint", len(got), len(got)-n)
+
+	// Old checkpoints go: two are kept.
+	for i := 0; i < 2; i++ {
+		time.Sleep(1100 * time.Millisecond) // a new name each second
+		if code, body := checkpointPost(t, s, checkpointToken); code != http.StatusOK {
+			t.Fatalf("checkpoint %d: %d %s", i+2, code, body)
+		}
+	}
+	entries, _ := os.ReadDir(dir)
+	var kept []string
+	for _, e := range entries {
+		kept = append(kept, filepath.Base(e.Name()))
+	}
+	if len(kept) != 2 {
+		t.Errorf("kept %v, want the newest two", kept)
+	}
+}
+
+func TestBackupCheckpointOff(t *testing.T) {
+	s := startServer(t)
+	if code, _ := checkpointPost(t, s, checkpointToken); code != http.StatusNotFound {
+		t.Errorf("checkpoints off: %d, want 404", code)
+	}
+}

server/e2e/helpers_test.go

@@ -91,3 +91,14 @@ func collectUnique(c *client, want int, overall, quiet time.Duration) (map[int]s
 	}
 	return got, dups, nil
 }
+
+// scaled scales a load size down under -race, where the server and the test
+// both run many times slower, so the suite stays within its time budget.
+func scaled(n int) int {
+	if raceEnabled {
+		if n /= 4; n < 1 {
+			n = 1
+		}
+	}
+	return n
+}

server/internal/checkpoint.go

@@ -0,0 +1,144 @@
+package internal
+
+import (
+	"crypto/subtle"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"os"
+	"path/filepath"
+	"sort"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/unit-io/unitdb/server/internal/pkg/log"
+	"github.com/unit-io/unitdb/server/internal/store"
+)
+
+// Checkpoints: a copy of the store that opens as it was at one moment
+// (store.Checkpoint), taken on request at POST /_checkpoint on the monitor
+// port, into CHECKPOINT_DIR, and kept there for a volume snapshot or a copy
+// elsewhere.
+// Restoring is starting a server with -db_path at the checkpoint, and the
+// same keyring.
+//
+// The monitor port has no other authentication, and a checkpoint holds
+// writes back for up to about 1.5 s, so:
+//   - it's off unless CHECKPOINT_DIR and CHECKPOINT_TOKEN are both set;
+//   - a request needs "Authorization: Bearer <CHECKPOINT_TOKEN>";
+//   - the server names the checkpoint's directory, never the request;
+//   - one runs at a time, and one per CHECKPOINT_MIN_MINUTES (10);
+//   - the newest CHECKPOINT_KEEP (3) are kept, older ones removed.
+
+const checkpointPrefix = "ckpt-"
+
+type checkpointer struct {
+	dir      string
+	token    string
+	keep     int
+	minEvery time.Duration
+
+	mu   sync.Mutex // held while one runs
+	last time.Time  // when the last one began, under mu
+
+	lastOK   atomic.Int64 // unix seconds of the last success; 0 before any
+	failures atomic.Int64
+	seconds  atomic.Int64 // milliseconds the last one took
+}
+
+// newCheckpointerFromEnv returns the checkpointer the environment asks for,
+// or nil when checkpoints are off.
+func newCheckpointerFromEnv() *checkpointer {
+	dir, token := os.Getenv("CHECKPOINT_DIR"), os.Getenv("CHECKPOINT_TOKEN")
+	if dir == "" || token == "" {
+		return nil
+	}
+	keep, _ := strconv.Atoi(os.Getenv("CHECKPOINT_KEEP"))
+	if keep < 1 {
+		keep = 3
+	}
+	mins, _ := strconv.Atoi(os.Getenv("CHECKPOINT_MIN_MINUTES"))
+	if mins < 0 || os.Getenv("CHECKPOINT_MIN_MINUTES") == "" {
+		mins = 10
+	}
+	return &checkpointer{dir: dir, token: token, keep: keep, minEvery: time.Duration(mins) * time.Minute}
+}
+
+func (c *checkpointer) authorized(r *http.Request) bool {
+	got := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")
+	return subtle.ConstantTimeCompare([]byte(got), []byte(c.token)) == 1
+}
+
+func (c *checkpointer) handle(w http.ResponseWriter, r *http.Request) {
+	if r.Method != http.MethodPost {
+		http.Error(w, "POST only", http.StatusMethodNotAllowed)
+		return
+	}
+	if !c.authorized(r) {
+		http.Error(w, "a checkpoint needs its token", http.StatusUnauthorized)
+		return
+	}
+	if !c.mu.TryLock() {
+		http.Error(w, "a checkpoint is running", http.StatusConflict)
+		return
+	}
+	defer c.mu.Unlock()
+	now := time.Now().UTC()
+	if !c.last.IsZero() && now.Sub(c.last) < c.minEvery {
+		http.Error(w, fmt.Sprintf("one checkpoint per %s", c.minEvery), http.StatusTooManyRequests)
+		return
+	}
+	c.last = now
+	dst := filepath.Join(c.dir, checkpointPrefix+now.Format("20060102T150405Z"))
+	start := time.Now()
+	err := store.Checkpoint(dst)
+	took := time.Since(start)
+	c.seconds.Store(took.Milliseconds())
+	if err != nil {
+		c.failures.Add(1)
+		os.RemoveAll(dst)
+		log.Error("checkpoint", err.Error())
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+	c.lastOK.Store(now.Unix())
+	log.Info("checkpoint", fmt.Sprintf("checkpoint %s in %s", dst, took.Round(time.Millisecond)))
+	c.prune()
+	w.Header().Set("content-type", "application/json")
+	json.NewEncoder(w).Encode(map[string]interface{}{
+		"dir":     dst,
+		"seconds": took.Seconds(),
+	})
+}
+
+// prune removes all but the newest keep checkpoints.
+func (c *checkpointer) prune() {
+	entries, err := os.ReadDir(c.dir)
+	if err != nil {
+		return
+	}
+	var names []string
+	for _, e := range entries {
+		if e.IsDir() && strings.HasPrefix(e.Name(), checkpointPrefix) {
+			names = append(names, e.Name())
+		}
+	}
+	sort.Strings(names) // the names sort by time
+	for len(names) > c.keep {
+		if err := os.RemoveAll(filepath.Join(c.dir, names[0])); err != nil {
+			log.Error("checkpoint", "removing an old checkpoint: "+err.Error())
+		}
+		names = names[1:]
+	}
+}
+
+func (c *checkpointer) writeMetrics(m *metricsWriter) {
+	m.one("unitdb_checkpoint_last_success_timestamp_seconds", "gauge",
+		"When the last checkpoint succeeded, in unix seconds; 0 before any.", float64(c.lastOK.Load()))
+	m.one("unitdb_checkpoint_failures_total", "counter", "Checkpoints that failed.", float64(c.failures.Load()))
+	m.one("unitdb_checkpoint_duration_seconds", "gauge",
+		"How long the last checkpoint took, writes held back meanwhile.", float64(c.seconds.Load())/1000)
+}

server/internal/db/adapter.go

@@ -88,4 +88,25 @@ type Adapter interface {
 	// Flush waits for the messages put before it to be written to the
 	// store's log, from which they are recovered after a crash.
 	Flush() error
+
+	// Checkpoint writes a copy of the store into dst, a directory that
+	// doesn't exist or is empty. The copy opens as the store was at one
+	// moment, as after a clean shutdown. Writes wait while it runs.
+	Checkpoint(dst string) error
+
+	// Stats returns the size of the store.
+	Stats() Stats
+}
+
+// Stats is the size of a store.
+type Stats struct {
+	// Messages is the number of messages in the store.
+	Messages uint64
+	// DiskBytes is the size of the store's files.
+	DiskBytes int64
+	// MemEntries is the number of records in memory (sessions, logs and the
+	// like), and MemSize the configured size of that memory (mem_size); 0
+	// if not configured.
+	MemEntries int64
+	MemSize    int64
 }