Compare commits

..

3 Commits

Author SHA1 Message Date
claude af4eeceb6a Keep the store's one connection, delete the seam it cannot survive (V-642)
`SetMaxOpenConns(1)` under WAL gives up concurrent reads, and the task
asked whether that costs anything. Measured over a fixed two-second
window, a paced writer against a read loop, three runs per cap:
reads do not queue. Four connections buy 70µs at p50 on a turn that
spends 1.19s in the resident model, and write throughput more than
halves. A 19ms worst case also cannot be the source of the 2.7s router
figure, so that line of enquiry is closed.

What the cap cannot survive is a long-lived transaction. It holds the
only connection, so a second read never completes: two seconds and
`context deadline exceeded`, against 1ms at a cap of four.

`Store.DB` handed out exactly that transaction. It had been there since
the initial commit with no production caller, and its comment described
a loop that never materialised. Its one user was a test helper reading
`delivery_attempts` by raw SQL, which `ListDeliveryAttempts` has covered
since V-390. So the cap stays and the seam goes, and the hazard is gone
by construction rather than by documentation.

`internal/store/conncap_test.go` stays as the standing measurement,
skipped under -short. The comment at the cap and the one in
`internal/ipc/server.go` that leans on it now state the invariant and
cite the numbers.

Measurement: docs/evals/2026-08-07-store-connection-cap.md

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-07 01:01:27 +04:00
kami 7b507dec94 Merge pull request 'factEnrichmentWorker walks the pending queue twice per tick to write one log line' (#191) from task/647-factenrichmentworker-walks-the-pending-q into master 2026-08-06 22:33:54 +02:00
claude 2c0334c4fe Count the enrichment backlog without a second query (V-647)
`tick` read `PendingFactResolutions` at the scan limit, then `status`
read it again with the same limit for one log line. Up to 2000 rows per
tick on a database that serialises reads, to say how long the queue is.

`statusOf` counts over a batch the caller already holds, and the tick
passes it the batch it just read. A resolved fact leaves the queue, so
the loop collects what is still pending rather than reporting the
pre-tick count. `status(ctx)` stays as the querying form, for a caller
outside the tick with no batch in hand.

No behaviour change: the three counts still describe one row set, and
the same facts are attempted per tick.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-07 00:32:54 +04:00
6 changed files with 277 additions and 28 deletions
+24 -8
View File
@@ -37,8 +37,8 @@ type factEnrichmentWorker struct {
nextTry map[int64]time.Time // fact id → earliest retry
}
// enrichmentScanLimit bounds how deep a single tick (or status report) walks
// the pending queue looking for facts whose backoff has elapsed. The queue is
// enrichmentScanLimit bounds how deep a single tick walks the pending queue
// looking for facts whose backoff has elapsed. The queue is
// ordered by id, so without a scan the oldest facts hold every batch slot
// whether or not they are eligible, and one permanently failing fact stalls
// every younger one behind it.
@@ -75,8 +75,8 @@ func newFactEnrichmentWorker(st *store.Store, eco *ecosystemWiring, interval tim
// has been down all day must be visible as a backlog, not as facts that
// silently never got tagged.
//
// All three numbers describe the same set of rows, the first
// enrichmentScanLimit pending facts. Counting Pending over a thousand rows
// All three numbers describe the same set of rows, whatever is still pending
// out of the first enrichmentScanLimit facts. Counting Pending over a thousand rows
// while counting InBackoff over the twenty that reached the head of a batch
// described two different populations under one struct.
type enrichmentStatus struct {
@@ -86,13 +86,22 @@ type enrichmentStatus struct {
Scanned int // rows the other three counts were taken over
}
// status reads the queue and counts over it. For a caller with no batch in
// hand — anything asking the worker how it is doing from outside the tick.
func (w *factEnrichmentWorker) status(ctx context.Context) enrichmentStatus {
var st enrichmentStatus
pending, err := w.store.PendingFactResolutions(ctx, enrichmentScanLimit)
if err != nil {
log.Printf("factenrichment: status: %v", err)
return st
return enrichmentStatus{}
}
return w.statusOf(pending)
}
// statusOf counts over a batch the caller already has. The batch is the query
// the tick already ran, so reporting the backlog costs no second read of the
// scan limit — up to a thousand rows, on a database that serialises them.
func (w *factEnrichmentWorker) statusOf(pending []store.Fact) enrichmentStatus {
var st enrichmentStatus
st.Pending = len(pending)
st.Scanned = len(pending)
w.mu.Lock()
@@ -144,17 +153,24 @@ func (w *factEnrichmentWorker) tick(ctx context.Context) {
}
w.forgetDeparted(pending)
skipped, failed, attempted := 0, 0, 0
// A resolved fact leaves the pending queue, so the batch in hand overstates
// the backlog by however many succeeded. Drop them here rather than
// re-reading the queue to find out.
remaining := make([]store.Fact, 0, len(pending))
for _, f := range pending {
if attempted >= w.batch {
break
remaining = append(remaining, f)
continue
}
if !w.due(f.ID) {
skipped++
remaining = append(remaining, f)
continue
}
attempted++
if !w.resolveOne(ctx, f) {
failed++
remaining = append(remaining, f)
}
}
if failed > 0 {
@@ -164,7 +180,7 @@ func (w *factEnrichmentWorker) tick(ctx context.Context) {
// Report the backlog every tick, not only when something failed: the
// stalled state worth seeing is the one where nothing failed because
// nothing was attempted.
if st := w.status(ctx); st.Pending > 0 {
if st := w.statusOf(remaining); st.Pending > 0 {
log.Printf("factenrichment: %d facts pending entity resolution, %d in backoff, worst attempt %d (scanned %d)",
st.Pending, st.InBackoff, st.MaxAttempts, st.Scanned)
}
@@ -0,0 +1,69 @@
# Does one sqlite connection make reads queue? No (V-642)
Measured 07-08-2026 at `7b507de`, on homesrv. The harness is
`internal/store/conncap_test.go`. It stays in the repo, because this claim gets
re-argued and the numbers should be re-runnable rather than quoted.
`internal/store/store.go` opens the database with `SetMaxOpenConns(1)`, while
`schema.sql` sets `journal_mode=WAL`. WAL exists to let readers run beside one
writer, so the cap gives up the thing the journal mode was chosen for. The
question was whether that costs anything.
## What was measured
A fixed two-second window. One writer calling `SetValue` paced at 2ms, and a
reader loop calling `RecentFacts(50)` over 500 seeded rows as fast as it can.
Same schema, same modernc driver, same machine, three runs per cap.
The window is wall-clock rather than a read count on purpose. A first version ran
a fixed 300 reads. That finished sooner at the higher cap, so it received fewer
writes, and two runs that did different work cannot be compared.
| cap | reads | writes | p50 | p95 | max |
|---|---|---|---|---|---|
| 1 | ~3050 | ~760 | 594µs | 900µs | 16-19ms |
| 4 | ~3600 | ~340 | 525µs | 710µs | 1-2ms |
## What it says
**Reads do not queue behind writes.** Four connections buy about 70µs at p50. A
turn spends 1.19s in the resident model. The tail does improve, from 19ms to 2ms,
and 19ms is still not a figure anyone notices in a spoken reply.
**Write throughput more than halves at the higher cap**, 760 writes against 340.
inference, not measured directly: at one connection the reader and the writer take
turns with no lock contention. At four the writer contends for the WAL write lock
with a live reader. Whatever the mechanism, the trade runs the opposite way from
the one the task expected.
**The cap was not the source of the 2.7s router figure.** CLAUDE.md records that
figure as contention rather than the model. This task was a candidate for where
that contention came from. A 19ms worst case cannot produce it. That line of
enquiry is closed.
**One transaction is what the cap cannot survive.** With a read-only transaction
open, a second read at cap 1 never completes. The harness gave it two seconds and
got `context deadline exceeded`. The same read at cap 4 took 1ms. The transaction
holds the only connection, so this is not a slow read, it is a stalled database.
## What was done
The cap stays at 1. The reason is now written where the cap is set, rather than
inferred from a four-word comment.
`Store.DB` was deleted. It handed out exactly the read-only transaction measured
above. It had been there since the initial commit with no production caller, and
its doc comment described a loop that never materialised. Its one user was a test
helper reading `delivery_attempts` by raw SQL. `ListDeliveryAttempts` has covered
that since V-390, and the helper now goes through the reader.
So the hazard is gone by construction, not by documentation.
`TestConnCap_ReadBlocksBehindOpenSnapshot` is the standing measurement of what
re-adding the seam would cost.
## Not answered
Whether reads queue on the deployed box under real load, as opposed to a
synthetic loop. The harness writes and reads one table. Digestion reads four and
embeds while it does. The finding that closes this task is the transaction stall,
which is structural and does not depend on load.
+11 -9
View File
@@ -94,20 +94,22 @@ func openTestStore(t *testing.T) *store.Store {
// attemptStatus reads one attempt row back. Returns ok=false when the row is
// gone, which would itself be a broken promise (a dropped attempt).
//
// It goes through ListDeliveryAttempts rather than raw SQL. This helper used to
// reach past the store into store.DB, which was the tell that the outbox was
// write-only; the reader landed in V-390 and this caller was not moved over.
func attemptStatus(t *testing.T, st *store.Store, id int64) (status string, completed bool, ok bool) {
t.Helper()
tx, err := st.DB(context.Background())
attempts, err := st.ListDeliveryAttempts(context.Background(), "", 200)
if err != nil {
t.Fatalf("read tx: %v", err)
t.Fatalf("ListDeliveryAttempts: %v", err)
}
defer func() { _ = tx.Rollback() }()
var completedTS *int64
err = tx.QueryRowContext(context.Background(),
`SELECT status, completed_ts FROM delivery_attempts WHERE id = ?`, id).Scan(&status, &completedTS)
if err != nil {
return "", false, false
for _, a := range attempts {
if a.ID == id {
return a.Status, a.HasComplete, true
}
}
return status, completedTS != nil, true
return "", false, false
}
// TestCrashBetweenBeginAndCompleteBecomesUnknown — simulate the crash window:
+5 -3
View File
@@ -17,9 +17,11 @@ import (
// Server — the core side of the boundary. Listens on a unix domain socket,
// accepts module connections, frames requests to a CoreAPI and responses back.
// One Server per daemon process; concurrent connections are handled in their
// own goroutine but share the single CoreAPI (and therefore the single store
// writer — store is single-connection, SetMaxOpenConns(1), so serialization is
// already guaranteed at the db; the Server adds no locking of its own).
// own goroutine but share the single CoreAPI, and so the single store writer.
// The store opens at SetMaxOpenConns(1), so serialisation is already guaranteed
// at the database and the Server adds no locking of its own. That cap is an
// invariant this comment depends on, measured and kept on 07-08-2026 (V-642,
// docs/evals/2026-08-07-store-connection-cap.md).
type Server struct {
api atomic.Value // stores CoreAPI
path string
+154
View File
@@ -0,0 +1,154 @@
package store
import (
"context"
"database/sql"
"fmt"
"path/filepath"
"sort"
"sync"
"sync/atomic"
"testing"
"time"
)
// conncap_test.go measures whether a read queues behind a write at
// SetMaxOpenConns(1), which is what openAt sets (V-642). It is a measurement
// harness, not an assertion: the numbers it prints are the evidence, and the
// decision to move the cap or leave it belongs in docs/evals.
//
// Run it with -v, and note that it is skipped under -short because it spends
// seconds on purpose.
// openCapped opens a plaintext store at the given connection cap. In-package,
// so it can reach the handle openAt caps at 1.
func openCapped(t *testing.T, cap int) *Store {
t.Helper()
path := filepath.Join(t.TempDir(), "cap.db")
db, err := openAt(context.Background(), path)
if err != nil {
t.Fatalf("openAt: %v", err)
}
db.SetMaxOpenConns(cap)
s := &Store{db: db}
t.Cleanup(func() { _ = s.Close() })
return s
}
func percentile(d []time.Duration, p float64) time.Duration {
if len(d) == 0 {
return 0
}
i := int(float64(len(d)-1) * p)
return d[i]
}
// seedFacts writes n facts so a read has rows to decode.
func seedFacts(t *testing.T, s *Store, n int) {
t.Helper()
ctx := context.Background()
now := time.Now().UTC()
for i := 0; i < n; i++ {
key := fmt.Sprintf("seed_%d", i)
if _, err := s.SetValue(ctx, KindSelf, key, "tap:test",
map[string]int{"ml": i}, now.Add(time.Duration(i)*time.Millisecond)); err != nil {
t.Fatalf("seed %d: %v", i, err)
}
}
}
// measureReadsUnderWrites reports read latency percentiles while a writer
// writes at a fixed pace. The pace matters: an unpaced writer completes a
// different number of writes at each cap, because at a higher cap it competes
// with the readers for the write lock instead of taking turns on one
// connection. Two runs that did different work cannot be compared.
// It runs for a fixed wall-clock window rather than a fixed read count, so the
// paced writer does the same work at every cap. Tying the window to a read
// count made the faster configuration receive fewer writes.
func measureReadsUnderWrites(t *testing.T, s *Store, window, pace time.Duration) []time.Duration {
t.Helper()
ctx := context.Background()
var stop atomic.Bool
var writes atomic.Int64
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
now := time.Now().UTC()
for i := 0; !stop.Load(); i++ {
key := fmt.Sprintf("hot_%d", i%16)
if _, err := s.SetValue(ctx, KindSelf, key, "tap:test",
map[string]int{"n": i}, now.Add(time.Duration(i)*time.Millisecond)); err != nil {
t.Errorf("write: %v", err)
return
}
writes.Add(1)
time.Sleep(pace)
}
}()
var lat []time.Duration
deadline := time.Now().Add(window)
for time.Now().Before(deadline) {
start := time.Now()
if _, err := s.RecentFacts(ctx, 50); err != nil {
t.Fatalf("RecentFacts: %v", err)
}
lat = append(lat, time.Since(start))
}
stop.Store(true)
wg.Wait()
t.Logf("in %v: %d reads, %d writes", window, len(lat), writes.Load())
sort.Slice(lat, func(i, j int) bool { return lat[i] < lat[j] })
return lat
}
// TestConnCap_ReadLatencyUnderWrites is the V-642 measurement: read latency at
// cap 1 against cap 4, same workload, same schema, same driver.
func TestConnCap_ReadLatencyUnderWrites(t *testing.T) {
if testing.Short() {
t.Skip("measurement harness; runs for seconds")
}
for _, cap := range []int{1, 4} {
t.Run(fmt.Sprintf("cap=%d", cap), func(t *testing.T) {
s := openCapped(t, cap)
seedFacts(t, s, 500)
lat := measureReadsUnderWrites(t, s, 2*time.Second, 2*time.Millisecond)
t.Logf("cap=%d reads=%d p50=%v p95=%v max=%v",
cap, len(lat), percentile(lat, 0.50), percentile(lat, 0.95), lat[len(lat)-1])
})
}
}
// TestConnCap_ReadBlocksBehindOpenSnapshot is the sharper claim: at cap 1 an
// open read-only transaction holds the only connection, so an unrelated read
// cannot proceed until it commits. This is why the store exposes no way to
// begin one — `Store.DB` used to, and was deleted in V-642 with no caller. The
// test stays as the reason, so re-adding that seam fails a measurement rather
// than shipping a stall.
func TestConnCap_ReadBlocksBehindOpenSnapshot(t *testing.T) {
if testing.Short() {
t.Skip("measurement harness; waits on a timeout")
}
for _, cap := range []int{1, 4} {
t.Run(fmt.Sprintf("cap=%d", cap), func(t *testing.T) {
s := openCapped(t, cap)
seedFacts(t, s, 50)
tx, err := s.db.BeginTx(context.Background(), &sql.TxOptions{ReadOnly: true})
if err != nil {
t.Fatalf("BeginTx: %v", err)
}
defer func() { _ = tx.Rollback() }()
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
start := time.Now()
_, err = s.RecentFacts(ctx, 10)
t.Logf("cap=%d read alongside an open snapshot: waited %v, err=%v",
cap, time.Since(start).Round(time.Millisecond), err)
})
}
}
+14 -8
View File
@@ -95,7 +95,20 @@ func openAt(ctx context.Context, path string) (*sql.DB, error) {
if err != nil {
return nil, fmt.Errorf("open %s: %w", path, err)
}
// single writer expected; the daemon is the only process touching the db.
// One connection, so every statement is serialised at the database and no
// caller above needs a lock of its own. internal/ipc's Server relies on
// exactly this, which is why the cap is an invariant rather than a tuning
// knob: raising it moves the serialisation guarantee somewhere it is not
// written down.
//
// Measured on 07-08-2026 (V-642, docs/evals/2026-08-07-store-connection-cap.md).
// WAL exists to let readers run beside one writer, and the cap gives that
// up, but reads do not queue: p50 594µs against 525µs at a cap of four,
// while write throughput more than halves. The one thing the cap cannot
// survive is a long-lived transaction, which holds the only connection and
// stalls every read for its lifetime. So the store begins none, and
// TestConnCap_ReadBlocksBehindOpenSnapshot is the standing measurement of
// what re-adding one would cost.
db.SetMaxOpenConns(1)
if _, err := db.ExecContext(ctx, schemaSQL); err != nil {
if closeErr := db.Close(); closeErr != nil {
@@ -133,13 +146,6 @@ func (s *Store) Close() error {
return s.enc.closeAndSeal(s.db)
}
// DB exposes the underlying handle for internal read-only snapshots.
// Used by the loop to take a consistent read under a single transaction.
// Modules never receive this handle — core mediates.
func (s *Store) DB(ctx context.Context) (*sql.Tx, error) {
return s.db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true})
}
var (
// ErrNoFact — no non-voided row exists for this key.
ErrNoFact = errors.New("store: no fact for key")