mirror of
https://github.com/hanzoai/replicate.git
synced 2026-10-30 18:23:20 +00:00
zeekay
a671dee75a
A version written into a manifest is written into every manifest that wants it, which is how eslint came to exist in a hundred and four forms across this estate and how a package came to resolve twice at run time. The catalog moves the number out of the manifest: a manifest names the package, the workspace names the version. Four are held below their latest because the newer release cannot be used, and the reason is recorded rather than the version alone: @noble/hashes 2 renamed every subpath the web3 stack imports, @splinetool/runtime 2 depends on a package that was never published, @dynamic-labs 5.4.1 the same, and react-resizable-panels 3 points its types condition at the bundle.
141 lines
3.4 KiB
Go
141 lines
3.4 KiB
Go
//go:build vfs && soak
|
|
// +build vfs,soak
|
|
|
|
package main_test
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"os"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
_ "github.com/hanzoai/sqlite"
|
|
|
|
"github.com/hanzoai/replicate/file"
|
|
"github.com/hanzoai/replicate/internal/testingutil"
|
|
)
|
|
|
|
// TestVFS_LongRunningSoak exercises the VFS under sustained read/write load.
|
|
// The default duration is 5 minutes but can be overridden with the
|
|
// REPLICATE_VFS_SOAK_DURATION environment variable (e.g. "10m").
|
|
func TestVFS_LongRunningSoak(t *testing.T) {
|
|
duration := 5 * time.Minute
|
|
if v := os.Getenv("REPLICATE_VFS_SOAK_DURATION"); v != "" {
|
|
if parsed, err := time.ParseDuration(v); err == nil {
|
|
duration = parsed
|
|
}
|
|
}
|
|
if testing.Short() && duration > time.Minute {
|
|
duration = time.Minute
|
|
}
|
|
|
|
client := file.NewReplicaClient(t.TempDir())
|
|
vfs := newVFS(t, client)
|
|
vfs.PollInterval = 100 * time.Millisecond
|
|
vfsName := registerTestVFS(t, vfs)
|
|
|
|
db, primary := openReplicatedPrimary(t, client, 75*time.Millisecond, 75*time.Millisecond)
|
|
defer testingutil.MustCloseSQLDB(t, primary)
|
|
|
|
if _, err := primary.Exec(`CREATE TABLE t (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
value TEXT,
|
|
updated_at INTEGER
|
|
)`); err != nil {
|
|
t.Fatalf("create table: %v", err)
|
|
}
|
|
seedLargeTable(t, primary, 1000)
|
|
forceReplicaSync(t, db)
|
|
|
|
replica := openVFSReplicaDB(t, vfsName)
|
|
defer replica.Close()
|
|
|
|
waitForReplicaRowCount(t, primary, replica, time.Minute)
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), duration)
|
|
defer cancel()
|
|
|
|
var writeOps atomic.Int64
|
|
var readOps atomic.Int64
|
|
errCh := make(chan error, 8)
|
|
var wg sync.WaitGroup
|
|
|
|
// Writers continuously mutate the primary database.
|
|
startWriter := func(name string) {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
rnd := time.NewTicker(7 * time.Millisecond)
|
|
defer rnd.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-rnd.C:
|
|
if _, err := primary.Exec("INSERT INTO t (value, updated_at) VALUES (?, strftime('%s','now'))", fmt.Sprintf("%s-%d", name, time.Now().UnixNano())); err != nil {
|
|
errCh <- fmt.Errorf("writer %s insert: %w", name, err)
|
|
return
|
|
}
|
|
if _, err := primary.Exec("UPDATE t SET value = value || '-w' WHERE id IN (SELECT id FROM t ORDER BY RANDOM() LIMIT 1)"); err != nil {
|
|
errCh <- fmt.Errorf("writer %s update: %w", name, err)
|
|
return
|
|
}
|
|
writeOps.Add(2)
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
startReader := func(name string) {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
default:
|
|
}
|
|
var minID, maxID, count int
|
|
if err := replica.QueryRow("SELECT IFNULL(MIN(id),0), IFNULL(MAX(id),0), COUNT(*) FROM t").Scan(&minID, &maxID, &count); err != nil {
|
|
errCh <- fmt.Errorf("reader %s query: %w", name, err)
|
|
return
|
|
}
|
|
if minID > maxID && count > 0 {
|
|
errCh <- fmt.Errorf("reader %s saw invalid range", name)
|
|
return
|
|
}
|
|
readOps.Add(1)
|
|
}
|
|
}()
|
|
}
|
|
|
|
for i := 0; i < 2; i++ {
|
|
startWriter(fmt.Sprintf("writer-%d", i))
|
|
}
|
|
for i := 0; i < 4; i++ {
|
|
startReader(fmt.Sprintf("reader-%d", i))
|
|
}
|
|
|
|
<-ctx.Done()
|
|
wg.Wait()
|
|
close(errCh)
|
|
for err := range errCh {
|
|
if err != nil {
|
|
t.Fatalf("soak error: %v", err)
|
|
}
|
|
}
|
|
|
|
if writeOps.Load() < int64(duration/time.Millisecond) {
|
|
t.Fatalf("expected sustained writes, got %d ops", writeOps.Load())
|
|
}
|
|
if readOps.Load() == 0 {
|
|
t.Fatalf("expected replica reads during soak")
|
|
}
|
|
|
|
waitForReplicaRowCount(t, primary, replica, time.Minute)
|
|
}
|