mirror of
https://github.com/hanzoai/replicate.git
synced 2026-10-30 18:23:20 +00:00
zeekay
a485f34a14
- All source files, configs, docs, CI renamed - Binary: replicate (not litestream) - Config: replicate.yml (litestream.yml as fallback) - Env: REPLICATE_* (LITESTREAM_* as fallback) - Socket: /var/run/replicate.sock - Image: ghcr.io/hanzoai/replicate - Module: github.com/hanzoai/replicate - Package name: replicate (Go), replicate-vfs (npm/pip/gem) - VFS: registered as "replicate", pragmas replicate_* - C exports: ReplicateVFSRegister, GoReplicate* - Windows service: Replicate - Systemd unit: replicate.service - Upstream attribution preserved (README, upgrade test scripts)
97 lines
2.2 KiB
Go
97 lines
2.2 KiB
Go
//go:build vfs && stress
|
|
// +build vfs,stress
|
|
|
|
package main_test
|
|
|
|
import (
|
|
"context"
|
|
"math/rand"
|
|
"os"
|
|
"runtime"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/hanzoai/replicate/file"
|
|
"github.com/hanzoai/replicate/internal/testingutil"
|
|
)
|
|
|
|
func TestVFS_RaceStressHarness(t *testing.T) {
|
|
if os.Getenv("REPLICATE_ALLOW_RACE") != "1" {
|
|
t.Skip("set REPLICATE_ALLOW_RACE=1 to run unstable race harness; modernc.org/sqlite checkptr panics are still unresolved")
|
|
}
|
|
if !runtime.RaceEnabled() {
|
|
t.Skip("requires go test -race")
|
|
}
|
|
|
|
client := file.NewReplicaClient(t.TempDir())
|
|
db, primary := openReplicatedPrimary(t, client, 20*time.Millisecond, 20*time.Millisecond)
|
|
defer testingutil.MustCloseSQLDB(t, primary)
|
|
|
|
if _, err := primary.Exec("CREATE TABLE stress (id INTEGER PRIMARY KEY, value TEXT)"); err != nil {
|
|
t.Fatalf("create table: %v", err)
|
|
}
|
|
seedLargeTable(t, primary, 100)
|
|
|
|
vfs := newVFS(t, client)
|
|
vfs.PollInterval = 5 * time.Millisecond
|
|
vfsName := registerTestVFS(t, vfs)
|
|
replica := openVFSReplicaDB(t, vfsName)
|
|
defer replica.Close()
|
|
waitForReplicaRowCount(t, primary, replica, 10*time.Second)
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
|
|
var writes atomic.Int64
|
|
go func() {
|
|
rnd := rand.New(rand.NewSource(time.Now().UnixNano()))
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
default:
|
|
}
|
|
if _, err := primary.Exec("INSERT INTO stress (value) VALUES (?)", randomPayload(rnd, 64)); err != nil && !isBusyError(err) {
|
|
t.Errorf("writer error: %v", err)
|
|
return
|
|
}
|
|
writes.Add(1)
|
|
}
|
|
}()
|
|
|
|
const readers = 64
|
|
errCh := make(chan error, readers)
|
|
for i := 0; i < readers; i++ {
|
|
go func() {
|
|
rnd := rand.New(rand.NewSource(time.Now().UnixNano()))
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
errCh <- nil
|
|
return
|
|
default:
|
|
}
|
|
var count int
|
|
if err := replica.QueryRow("SELECT COUNT(*) FROM stress WHERE id >= ?", rnd.Intn(50)).Scan(&count); err != nil {
|
|
if isBusyError(err) {
|
|
continue
|
|
}
|
|
errCh <- err
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
for i := 0; i < readers; i++ {
|
|
if err := <-errCh; err != nil {
|
|
t.Fatalf("reader error: %v", err)
|
|
}
|
|
}
|
|
|
|
if writes.Load() == 0 {
|
|
t.Fatalf("writer never made progress")
|
|
}
|
|
}
|