//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)
}