Skip to content

Commit 92806e5

Browse files
committed
fix: single write connection prevents flaky notification tests
The write pool defaulted to unlimited connections, but powersync_update_hooks state is per-connection. install and get could hit different connections, so get always returned []. SetMaxOpenConns(1) on the write pool fixes this — SQLite only allows one writer anyway. Also kick the watcher immediately from Exec and sqlc write queries so subscribers don't wait for the 50ms poll tick.
1 parent fb6d116 commit 92806e5

2 files changed

Lines changed: 61 additions & 4 deletions

File tree

‎internal/sqlite/database.go‎

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -150,6 +150,10 @@ func Open(ctx context.Context, path string) (DB, error) {
150150
db.Close()
151151
return nil, fmt.Errorf("ping write pool: %w", err)
152152
}
153+
// SQLite only allows one writer at a time. A single connection avoids
154+
// SQLITE_BUSY between writers and ensures powersync_update_hooks state
155+
// (which is per-connection) is shared across all write operations.
156+
db.SetMaxOpenConns(1)
153157

154158
// Database-level pragmas (not per-connection — one exec is correct).
155159
// WAL allows concurrent readers during writes — the core fix for query blocking during sync.
@@ -236,8 +240,9 @@ func (d *database) ReadQueries() *gen.Queries {
236240
}
237241

238242
// WriteQueries returns a Queries instance backed by the write pool.
243+
// Writes kick the watcher so subscribers are notified immediately.
239244
func (d *database) WriteQueries() *gen.Queries {
240-
return gen.New(d.db)
245+
return gen.New(&kickingDB{db: d.db, kick: d.kickWatcher})
241246
}
242247

243248
// ---------------------------------------------------------------------------
@@ -295,9 +300,14 @@ func (d *database) QueryRow(ctx context.Context, query string, args ...any) *sql
295300
return d.readDB.QueryRowContext(ctx, query, args...)
296301
}
297302

298-
// Exec executes a statement via the write pool.
303+
// Exec executes a statement via the write pool and kicks the watcher so
304+
// subscribers are notified without waiting for the next poll tick.
299305
func (d *database) Exec(ctx context.Context, query string, args ...any) (sql.Result, error) {
300-
return d.db.ExecContext(ctx, query, args...)
306+
result, err := d.db.ExecContext(ctx, query, args...)
307+
if err == nil {
308+
d.kickWatcher()
309+
}
310+
return result, err
301311
}
302312

303313
// BeginTx starts a transaction on the write pool.

‎internal/sqlite/watch.go‎

Lines changed: 48 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,10 +2,13 @@ package sqlite
22

33
import (
44
"context"
5+
"database/sql"
56
"encoding/json"
67
"strings"
78
"sync"
89
"time"
10+
11+
"github.com/usetero/cli/internal/sqlite/gen"
912
)
1013

1114
const (
@@ -32,12 +35,39 @@ func (s *Subscription) Stop() {
3235
s.db.unsubscribe(s)
3336
}
3437

38+
// kickingDB wraps a DBTX and kicks the watcher after successful writes.
39+
type kickingDB struct {
40+
db gen.DBTX
41+
kick func()
42+
}
43+
44+
func (k *kickingDB) ExecContext(ctx context.Context, query string, args ...interface{}) (sql.Result, error) {
45+
result, err := k.db.ExecContext(ctx, query, args...)
46+
if err == nil {
47+
k.kick()
48+
}
49+
return result, err
50+
}
51+
52+
func (k *kickingDB) PrepareContext(ctx context.Context, query string) (*sql.Stmt, error) {
53+
return k.db.PrepareContext(ctx, query)
54+
}
55+
56+
func (k *kickingDB) QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error) {
57+
return k.db.QueryContext(ctx, query, args...)
58+
}
59+
60+
func (k *kickingDB) QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row {
61+
return k.db.QueryRowContext(ctx, query, args...)
62+
}
63+
3564
// watchState manages change detection and subscriber notification.
3665
type watchState struct {
3766
mu sync.RWMutex
3867
subscribers map[*Subscription]struct{}
3968
installed bool
4069
cancel context.CancelFunc
70+
kick chan struct{} // signals an immediate check
4171
}
4272

4373
// installUpdateHooks enables change tracking via PowerSync's update hooks.
@@ -60,13 +90,15 @@ func (d *database) installUpdateHooks(ctx context.Context) error {
6090
// Start background polling goroutine
6191
pollCtx, cancel := context.WithCancel(ctx)
6292
d.watch.cancel = cancel
93+
d.watch.kick = make(chan struct{}, 1)
6394
go d.pollForChanges(pollCtx)
6495

6596
d.watch.installed = true
6697
return nil
6798
}
6899

69-
// pollForChanges runs in the background, checking for table changes periodically.
100+
// pollForChanges runs in the background, checking for table changes periodically
101+
// or immediately when kicked by Exec.
70102
func (d *database) pollForChanges(ctx context.Context) {
71103
ticker := time.NewTicker(pollInterval)
72104
defer ticker.Stop()
@@ -77,10 +109,25 @@ func (d *database) pollForChanges(ctx context.Context) {
77109
return
78110
case <-ticker.C:
79111
d.checkForChanges()
112+
case <-d.watch.kick:
113+
d.checkForChanges()
80114
}
81115
}
82116
}
83117

118+
// kickWatcher signals the poll loop to check immediately.
119+
// Safe to call before hooks are installed (no-op).
120+
func (d *database) kickWatcher() {
121+
if d.watch.kick == nil {
122+
return
123+
}
124+
select {
125+
case d.watch.kick <- struct{}{}:
126+
default:
127+
// Already pending — poller will pick it up.
128+
}
129+
}
130+
84131
// Subscribe returns a Subscription that receives table names when they change.
85132
// The channel has a buffer of 1 to avoid blocking the poller.
86133
// Call Stop() on the subscription when done to prevent leaks.

0 commit comments

Comments
 (0)