feat: E2E encrypted streaming replication to S3 via luxfi/age

- replicate.go: Replicator with Start/Stop, Incremental, Snapshot, Restore - ZAP binary format (NOT protobuf) — .zap/.zap.age extensions - Fixed stale "protobuf-encoded" comments in backup.go - Age encryption (X25519, upgradeable to X-Wing/ML-KEM-768) - S3 via minio-go/v7 (works with hanzoai/s3, AWS, GCS, R2) - S3 path: {prefix}/{service}/{pod}/{version}.zap.age - E2E test: 100 keys → full backup → 50 more → incremental → encrypt → restore → verify 150 keys - 6 tests, all pass Compatible with hanzoai/replicate (SQLite) for the full encryption stack.
2026-07-26 22:46:36 +00:00 · 2026-04-09 17:46:23 -07:00
parent a9f86e5805
commit 9a96d340a1
5 changed files with 251 additions and 13 deletions
@@ -24,7 +24,7 @@ import (
 // than the maxBatchSize.
 const flushThreshold = 100 << 20

-// Backup dumps a protobuf-encoded list of all entries in the database into the
+// Backup dumps a ZAP binary-encoded list of all entries in the database into the
 // given writer, that are newer than or equal to the specified version. It
 // returns a timestamp (version) indicating the version of last entry that is
 // dumped, which after incrementing by 1 can be passed into later invocation to
@@ -41,7 +41,7 @@ func (db *DB) Backup(w io.Writer, since uint64) (uint64, error) {
 	return stream.Backup(w, since)
 }

-// Backup dumps a protobuf-encoded list of all entries in the database into the
+// Backup dumps a ZAP binary-encoded list of all entries in the database into the
 // given writer, that are newer than or equal to the specified version. It returns a
 // timestamp(version) indicating the version of last entry that was dumped, which
 // after incrementing by 1 can be passed into a later invocation to generate an
@@ -221,7 +221,7 @@ func (l *KVLoader) Finish() error {
 	return l.throttle.Finish()
 }

-// Load reads a protobuf-encoded list of all entries from a reader and writes
+// Load reads a ZAP binary-encoded list of all entries from a reader and writes
 // them to the database. This can be used to restore the database from a backup
 // made by calling DB.Backup(). If more complex logic is needed to restore a badger
 // backup, the KVLoader interface should be used instead.
@@ -0,0 +1,107 @@
+/* Code generated by cmd/cgo; DO NOT EDIT. */
+
+/* package github.com/luxfi/zapdb/v4/bindings/cabi */
+
+
+#line 1 "cgo-builtin-export-prolog"
+
+#include <stddef.h>
+
+#ifndef GO_CGO_EXPORT_PROLOGUE_H
+#define GO_CGO_EXPORT_PROLOGUE_H
+
+#ifndef GO_CGO_GOSTRING_TYPEDEF
+typedef struct { const char *p; ptrdiff_t n; } _GoString_;
+extern size_t _GoStringLen(_GoString_ s);
+extern const char *_GoStringPtr(_GoString_ s);
+#endif
+
+#endif
+
+/* Start of preamble from import "C" comments.  */
+
+
+#line 16 "main.go"
+ #include <string.h>
+
+#line 1 "cgo-generated-wrapper"
+
+
+/* End of preamble from import "C" comments.  */
+
+
+/* Start of boilerplate cgo prologue.  */
+#line 1 "cgo-gcc-export-header-prolog"
+
+#ifndef GO_CGO_PROLOGUE_H
+#define GO_CGO_PROLOGUE_H
+
+typedef signed char GoInt8;
+typedef unsigned char GoUint8;
+typedef short GoInt16;
+typedef unsigned short GoUint16;
+typedef int GoInt32;
+typedef unsigned int GoUint32;
+typedef long long GoInt64;
+typedef unsigned long long GoUint64;
+typedef GoInt64 GoInt;
+typedef GoUint64 GoUint;
+typedef size_t GoUintptr;
+typedef float GoFloat32;
+typedef double GoFloat64;
+#ifdef _MSC_VER
+#if !defined(__cplusplus) || _MSVC_LANG <= 201402L
+#include <complex.h>
+typedef _Fcomplex GoComplex64;
+typedef _Dcomplex GoComplex128;
+#else
+#include <complex>
+typedef std::complex<float> GoComplex64;
+typedef std::complex<double> GoComplex128;
+#endif
+#else
+typedef float _Complex GoComplex64;
+typedef double _Complex GoComplex128;
+#endif
+
+/*
+  static assertion to make sure the file is being used on architecture
+  at least with matching size of GoInt.
+*/
+typedef char _check_for_64_bit_pointer_matching_GoInt[sizeof(void*)==64/8 ? 1:-1];
+
+#ifndef GO_CGO_GOSTRING_TYPEDEF
+typedef _GoString_ GoString;
+#endif
+typedef void *GoMap;
+typedef void *GoChan;
+typedef struct { void *t; void *v; } GoInterface;
+typedef struct { void *data; GoInt len; GoInt cap; } GoSlice;
+
+#endif
+
+/* End of boilerplate cgo prologue.  */
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+extern int lux_zapdb_open(char* path, int pathLen, unsigned long long* out);
+extern int lux_zapdb_close(unsigned long long handle);
+extern int lux_zapdb_get(unsigned long long handle, char* key, int keyLen, char* val, int* valLen);
+extern int lux_zapdb_set(unsigned long long handle, char* key, int keyLen, char* val, int valLen);
+extern int lux_zapdb_delete(unsigned long long handle, char* key, int keyLen);
+extern int lux_zapdb_iterate(unsigned long long handle, char* prefix, int prefixLen, void* callback);
+extern int lux_zapdb_backup(unsigned long long handle, char* path, int pathLen);
+extern int lux_zapdb_load(unsigned long long handle, char* path, int pathLen);
+extern int lux_zapdb_sync(unsigned long long handle);
+extern int lux_zapdb_compact(unsigned long long handle);
+extern int lux_zapdb_new_txn(unsigned long long handle, int readOnly, unsigned long long* out);
+extern int lux_zapdb_txn_commit(unsigned long long txnHandle);
+extern void lux_zapdb_txn_discard(unsigned long long txnHandle);
+extern int lux_zapdb_txn_get(unsigned long long txnHandle, char* key, int keyLen, char* val, int* valLen);
+extern int lux_zapdb_txn_set(unsigned long long txnHandle, char* key, int keyLen, char* val, int valLen);
+
+#ifdef __cplusplus
+}
+#endif
@@ -308,17 +308,17 @@ func (r *Replicator) downloadAndLoad(ctx context.Context, key string) error {
 }

 func (r *Replicator) incKey(version uint64) string {
-	ext := ".pb"
+	ext := ".zap"
 	if r.cfg.AgeRecipient != nil {
-		ext = ".pb.age"
+		ext = ".zap.age"
 	}
 	return path.Join(r.cfg.Path, "inc", fmt.Sprintf("%020d%s", version, ext))
 }

 func (r *Replicator) snapKey(t time.Time) string {
-	ext := ".pb"
+	ext := ".zap"
 	if r.cfg.AgeRecipient != nil {
-		ext = ".pb.age"
+		ext = ".zap.age"
 	}
 	return path.Join(r.cfg.Path, "snap", fmt.Sprintf("%d%s", t.UnixNano(), ext))
 }
@@ -326,9 +326,9 @@ func (r *Replicator) snapKey(t time.Time) string {
 // versionFromKey extracts the version number from an incremental backup key.
 func versionFromKey(key string) uint64 {
 	base := path.Base(key)
-	// Strip extensions (.pb, .pb.age)
+	// Strip extensions (.zap, .zap.age)
 	base = strings.TrimSuffix(base, ".age")
-	base = strings.TrimSuffix(base, ".pb")
+	base = strings.TrimSuffix(base, ".zap")
 	v, _ := strconv.ParseUint(base, 10, 64)
 	return v
 }
@@ -228,10 +228,10 @@ func TestReplicatorVersionFromKey(t *testing.T) {
 		key  string
 		want uint64
 	}{
-		{"zapdb/node-0/inc/00000000000000000042.pb", 42},
-		{"zapdb/node-0/inc/00000000000000000042.pb.age", 42},
-		{"zapdb/node-0/inc/00000000000000001000.pb", 1000},
-		{"some/path/inc/00000000000000000001.pb.age", 1},
+		{"zapdb/node-0/inc/00000000000000000042.zap", 42},
+		{"zapdb/node-0/inc/00000000000000000042.zap.age", 42},
+		{"zapdb/node-0/inc/00000000000000001000.zap", 1000},
+		{"some/path/inc/00000000000000000001.zap.age", 1},
 	}
 	for _, tt := range tests {
 		got := versionFromKey(tt.key)
@@ -293,3 +293,134 @@ func TestReplicatorEncryptedIncremental(t *testing.T) {
 	require.NoError(t, db2.Close())
 	_ = maxV
 }
+
+// TestReplicateE2E is the full end-to-end test:
+//   1. Create source DB, write 100 keys
+//   2. Full backup → age encrypt → buffer (simulating S3)
+//   3. Write 50 more keys to source
+//   4. Incremental backup → age encrypt → buffer
+//   5. Restore to fresh DB: decrypt full → load, decrypt inc → load
+//   6. Verify ALL 150 keys match
+//   7. Verify encrypted blobs cannot be loaded without decryption
+func TestReplicateE2E(t *testing.T) {
+	identity, err := age.GenerateX25519Identity()
+	require.NoError(t, err)
+	recipient := identity.Recipient()
+
+	// ── Source DB ──
+	srcDir, err := os.MkdirTemp("", "zapdb-e2e-src")
+	require.NoError(t, err)
+	defer removeDir(srcDir)
+
+	src, err := Open(getTestOptions(srcDir))
+	require.NoError(t, err)
+
+	// Write batch 1: 100 keys.
+	for i := 0; i < 100; i++ {
+		k := []byte(fmt.Sprintf("user:%06d", i))
+		v := []byte(fmt.Sprintf(`{"id":%d,"name":"user-%d","balance":%d}`, i, i, (i+1)*1000))
+		require.NoError(t, src.Update(func(txn *Txn) error {
+			return txn.Set(k, v)
+		}))
+	}
+
+	// ── Full backup + encrypt ──
+	var fullPlain bytes.Buffer
+	sinceV, err := src.Backup(&fullPlain, 0)
+	require.NoError(t, err)
+	require.True(t, sinceV > 0, "sinceV should be > 0")
+	t.Logf("Full backup: %d bytes, maxVersion=%d", fullPlain.Len(), sinceV)
+
+	fullEnc, err := encrypt(&fullPlain, recipient)
+	require.NoError(t, err)
+	t.Logf("Full encrypted: %d bytes", fullEnc.Len())
+
+	// ── Write batch 2: 50 more keys ──
+	for i := 100; i < 150; i++ {
+		k := []byte(fmt.Sprintf("user:%06d", i))
+		v := []byte(fmt.Sprintf(`{"id":%d,"name":"user-%d","balance":%d}`, i, i, (i+1)*1000))
+		require.NoError(t, src.Update(func(txn *Txn) error {
+			return txn.Set(k, v)
+		}))
+	}
+
+	// Also update an existing key to test overwrites.
+	require.NoError(t, src.Update(func(txn *Txn) error {
+		return txn.Set([]byte("user:000000"), []byte(`{"id":0,"name":"user-0-UPDATED","balance":999999}`))
+	}))
+
+	// ── Incremental backup + encrypt ──
+	var incPlain bytes.Buffer
+	sinceV2, err := src.Backup(&incPlain, sinceV)
+	require.NoError(t, err)
+	require.True(t, sinceV2 > sinceV, "sinceV2=%d should > sinceV=%d", sinceV2, sinceV)
+	require.True(t, incPlain.Len() > 0, "incremental should have data")
+	t.Logf("Incremental backup: %d bytes, maxVersion=%d", incPlain.Len(), sinceV2)
+
+	incEnc, err := encrypt(&incPlain, recipient)
+	require.NoError(t, err)
+	t.Logf("Incremental encrypted: %d bytes", incEnc.Len())
+
+	require.NoError(t, src.Close())
+
+	// ── Verify encrypted data is not directly loadable ──
+	// (age header bytes aren't valid ZAP framing — Load panics on bad length prefix)
+	t.Run("encrypted_not_loadable", func(t *testing.T) {
+		// Load panics on corrupt framing — verify it doesn't silently succeed.
+		require.Panics(t, func() {
+			badDir, _ := os.MkdirTemp("", "zapdb-e2e-bad")
+			defer removeDir(badDir)
+			badDB, _ := Open(getTestOptions(badDir))
+			defer badDB.Close()
+			_ = badDB.Load(bytes.NewReader(fullEnc.Bytes()), 16)
+		}, "loading encrypted data without decryption should panic on bad framing")
+	})
+
+	// ── Restore: decrypt + load ──
+	dstDir, err := os.MkdirTemp("", "zapdb-e2e-dst")
+	require.NoError(t, err)
+	defer removeDir(dstDir)
+
+	dst, err := Open(getTestOptions(dstDir))
+	require.NoError(t, err)
+
+	// Decrypt and load full backup.
+	fullDec, err := age.Decrypt(fullEnc, identity)
+	require.NoError(t, err)
+	fullDecBytes, err := io.ReadAll(fullDec)
+	require.NoError(t, err)
+	require.NoError(t, dst.Load(bytes.NewReader(fullDecBytes), 16))
+
+	// Decrypt and load incremental.
+	incDec, err := age.Decrypt(incEnc, identity)
+	require.NoError(t, err)
+	incDecBytes, err := io.ReadAll(incDec)
+	require.NoError(t, err)
+	require.NoError(t, dst.Load(bytes.NewReader(incDecBytes), 16))
+
+	// ── Verify ALL 150 keys ──
+	require.NoError(t, dst.View(func(txn *Txn) error {
+		for i := 0; i < 150; i++ {
+			k := []byte(fmt.Sprintf("user:%06d", i))
+			item, err := txn.Get(k)
+			if err != nil {
+				return fmt.Errorf("missing key %s: %w", k, err)
+			}
+			err = item.Value(func(val []byte) error {
+				if i == 0 {
+					// This was updated in the incremental.
+					require.Contains(t, string(val), "UPDATED", "user:000000 should be updated")
+				}
+				require.Contains(t, string(val), fmt.Sprintf(`"id":%d`, i))
+				return nil
+			})
+			if err != nil {
+				return err
+			}
+		}
+		return nil
+	}))
+
+	t.Logf("All 150 keys verified, including overwrite of user:000000")
+	require.NoError(t, dst.Close())
+}