//go:build spike // Phase-0 de-risk spike. This is the whole point of Phase 0: prove that the // pinned github.com/dolthub/dolt/go module can init a bare NBS store, serve it // over remotesrv, and round-trip a real `dolt` CLI (v2.1.10) clone -> insert -> // commit -> push -> re-clone. If this passes, the storage format and remotesapi // RPCs are compatible between the module and the CLI, and the parallel waves // can proceed. // // Run with: // // CGO_CPPFLAGS=-I/opt/homebrew/opt/icu4c@78/include \ // CGO_LDFLAGS=-L/opt/homebrew/opt/icu4c@78/lib \ // go test -tags spike ./storage/ -run TestSpike -v // // The test skips (does not fail) when the dolt CLI is absent. package storage import ( "context" "fmt" "net" "os" "os/exec" "path/filepath" "strings" "sync" "testing" "time" remotesapi "github.com/dolthub/dolt/go/gen/proto/dolt/services/remotesapi/v1alpha1" "github.com/dolthub/dolt/go/libraries/doltcore/doltdb" "github.com/dolthub/dolt/go/libraries/doltcore/remotesrv" "github.com/dolthub/dolt/go/libraries/utils/earl" "github.com/dolthub/dolt/go/libraries/utils/filesys" "github.com/dolthub/dolt/go/store/nbs" "github.com/dolthub/dolt/go/store/types" "github.com/sirupsen/logrus" ) const doltBin = "/opt/homebrew/bin/dolt" // spikeCache is a minimal remotesrv.DBCache modeled on the upstream // utils/remotesrv LocalCSCache: it memoizes one nbs.NewLocalStore per repo // path, resolved under a fixed root directory. Real storage/ (wave A) will // grow repo-row validation and eviction on top of this shape. type spikeCache struct { mu sync.Mutex root string dbs map[string]remotesrv.RemoteSrvStore } func (c *spikeCache) Get(ctx context.Context, path, nbfVerStr string) (remotesrv.RemoteSrvStore, error) { c.mu.Lock() defer c.mu.Unlock() id := filepath.FromSlash(strings.Trim(path, "/")) if cs, ok := c.dbs[id]; ok { return cs, nil } abs := filepath.Join(c.root, id) if err := os.MkdirAll(abs, 0o755); err != nil { return nil, err } cs, err := nbs.NewLocalStore(ctx, nbfVerStr, abs, 128*1024*1024, nbs.NewUnlimitedMemQuotaProvider(), false) if err != nil { return nil, err } c.dbs[id] = cs return cs, nil } func TestSpike(t *testing.T) { if _, err := os.Stat(doltBin); err != nil { t.Skipf("dolt CLI not found at %s (%v); skipping interop spike", doltBin, err) } ctx := context.Background() root := t.TempDir() reposRoot := filepath.Join(root, "repos") // 1. Create a bare NBS store at repos/test/db and write an empty repo. storeDir := filepath.Join(reposRoot, "test", "db") if err := os.MkdirAll(storeDir, 0o755); err != nil { t.Fatal(err) } fileURL := earl.FileUrlFromPath(storeDir, os.PathSeparator) ddb, err := doltdb.LoadDoltDB(ctx, types.Format_DOLT, fileURL, filesys.LocalFS) if err != nil { t.Fatalf("LoadDoltDB(%s): %v", fileURL, err) } if err := ddb.WriteEmptyRepo(ctx, "main", "spike", "spike@test.local"); err != nil { t.Fatalf("WriteEmptyRepo: %v", err) } // Release the init handle before the server opens its own store over the // same directory. if err := ddb.Close(); err != nil { t.Fatalf("close init DoltDB: %v", err) } // 2. Serve repos/ over remotesrv on a single ephemeral localhost port // (http + gRPC multiplexed), no auth interceptors, plain http. addr := freeAddr(t) cache := &spikeCache{root: reposRoot, dbs: map[string]remotesrv.RemoteSrvStore{}} logger := logrus.New() logger.SetLevel(logrus.ErrorLevel) // The server FS must be rooted at reposRoot so chunk-download URL prefixes // are clean relatives (e.g. "test/db") rather than "../../.." paths, which // the sealed-URL file handler rejects. Upstream achieves this by chdir-ing // to the repos root; we set the FS working dir instead to avoid mutating // global process state. fs, err := filesys.LocalFilesysWithWorkingDir(reposRoot) if err != nil { t.Fatalf("LocalFilesysWithWorkingDir(%s): %v", reposRoot, err) } server, err := remotesrv.NewServer(remotesrv.ServerArgs{ Logger: logrus.NewEntry(logger), HttpHost: "", // echo request :authority into chunk URLs HttpListenAddr: addr, GrpcListenAddr: addr, // == HttpListenAddr => single-port multiplex FS: fs, DBCache: cache, ConcurrencyControl: remotesapi.PushConcurrencyControl_PUSH_CONCURRENCY_CONTROL_IGNORE_WORKING_SET, }) if err != nil { t.Fatalf("remotesrv.NewServer: %v", err) } listeners, err := server.Listeners() if err != nil { t.Fatalf("server.Listeners: %v", err) } go server.Serve(listeners) defer server.GracefulStop() waitTCP(t, addr) // 3. Drive the real dolt CLI against the server in an isolated HOME so the // developer's real dolt config is untouched. home := filepath.Join(root, "home") if err := os.MkdirAll(home, 0o755); err != nil { t.Fatal(err) } env := append(os.Environ(), "HOME="+home) runDolt(t, env, root, "config", "--global", "--add", "user.email", "spike@test.local") runDolt(t, env, root, "config", "--global", "--add", "user.name", "spike") url := fmt.Sprintf("http://%s/test/db", addr) clone1 := filepath.Join(root, "clone1") if err := os.MkdirAll(clone1, 0o755); err != nil { t.Fatal(err) } runDolt(t, env, clone1, "clone", url) work := filepath.Join(clone1, "db") runDolt(t, env, work, "sql", "-q", "create table t(i int primary key); insert into t values (1),(2),(3);") runDolt(t, env, work, "commit", "-Am", "spike commit") runDolt(t, env, work, "push", "origin", "main") // 4. Fresh re-clone and verify the pushed rows are present. clone2 := filepath.Join(root, "clone2") if err := os.MkdirAll(clone2, 0o755); err != nil { t.Fatal(err) } runDolt(t, env, clone2, "clone", url) out := runDolt(t, env, filepath.Join(clone2, "db"), "sql", "-q", "select i from t order by i", "-r", "csv") for _, want := range []string{"1", "2", "3"} { if !strings.Contains(out, want) { t.Fatalf("re-cloned db missing row %q; got csv:\n%s", want, out) } } t.Logf("spike round-trip OK; re-clone csv:\n%s", out) } // freeAddr returns a currently-free 127.0.0.1 address. There is an inherent // race between closing the probe listener and the server binding, but it is // acceptable for a single local test. func freeAddr(t *testing.T) string { t.Helper() l, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { t.Fatalf("probe listen: %v", err) } addr := l.Addr().String() if err := l.Close(); err != nil { t.Fatalf("probe close: %v", err) } return addr } // waitTCP blocks until addr accepts a TCP connection or the deadline elapses. func waitTCP(t *testing.T, addr string) { t.Helper() deadline := time.Now().Add(10 * time.Second) for time.Now().Before(deadline) { conn, err := net.DialTimeout("tcp", addr, 200*time.Millisecond) if err == nil { conn.Close() return } time.Sleep(50 * time.Millisecond) } t.Fatalf("server at %s never became reachable", addr) } // runDolt runs the dolt CLI in dir with env, fails the test on non-zero exit, // and returns combined stdout+stderr. func runDolt(t *testing.T, env []string, dir string, args ...string) string { t.Helper() cmd := exec.Command(doltBin, args...) cmd.Dir = dir cmd.Env = env out, err := cmd.CombinedOutput() if err != nil { t.Fatalf("dolt %s failed: %v\n%s", strings.Join(args, " "), err, out) } return string(out) }