Skip to content

Commit 39480d1

Browse files
committed
refactor: remove watcher/subscriber infrastructure
Consumers (catalogstatus, policystatus) poll on their own tick intervals and never used Subscribe(). The watcher relied on powersync_update_hooks which has per-connection state — fundamentally broken with connection pools. SetMaxOpenConns(1) fixed the tests but caused a sync deadlock (Controller holds the sole write connection, starving the uploader). Remove the whole thing: watch.go, watch_test.go, kickingDB, watchState, installUpdateHooks, Subscribe, SetMaxOpenConns(1).
1 parent 92806e5 commit 39480d1

5 files changed

Lines changed: 5 additions & 758 deletions

File tree

‎internal/sqlite/database.go‎

Lines changed: 5 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,6 @@ type DB interface {
4444

4545
// Sync
4646
PendingUploadCounts(ctx context.Context) (map[Table]int64, error)
47-
Subscribe() *Subscription
4847

4948
// Low-level
5049
Query(ctx context.Context, sql string, args ...any) (*sql.Rows, error)
@@ -61,7 +60,6 @@ type database struct {
6160
db *sql.DB // write pool — WAL writer, PowerSync controller, mutations
6261
readDB *sql.DB // read pool — query_only, all domain reads
6362
path string
64-
watch watchState
6563
}
6664

6765
// Ensure database implements DB.
@@ -150,11 +148,6 @@ func Open(ctx context.Context, path string) (DB, error) {
150148
db.Close()
151149
return nil, fmt.Errorf("ping write pool: %w", err)
152150
}
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)
157-
158151
// Database-level pragmas (not per-connection — one exec is correct).
159152
// WAL allows concurrent readers during writes — the core fix for query blocking during sync.
160153
// journal_size_limit caps the WAL file at 6MB to prevent unbounded growth.
@@ -181,22 +174,11 @@ func Open(ctx context.Context, path string) (DB, error) {
181174
return nil, fmt.Errorf("ping read pool: %w", err)
182175
}
183176

184-
d := &database{
177+
return &database{
185178
db: db,
186179
readDB: readDB,
187180
path: path,
188-
}
189-
190-
// Install update hooks for change notifications (only works with PowerSync extension)
191-
if extensionPath != "" {
192-
if err := d.installUpdateHooks(ctx); err != nil {
193-
db.Close()
194-
readDB.Close()
195-
return nil, fmt.Errorf("install update hooks: %w", err)
196-
}
197-
}
198-
199-
return d, nil
181+
}, nil
200182
}
201183

202184
// Close closes both the read and write connection pools.
@@ -240,9 +222,8 @@ func (d *database) ReadQueries() *gen.Queries {
240222
}
241223

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

248229
// ---------------------------------------------------------------------------
@@ -300,14 +281,9 @@ func (d *database) QueryRow(ctx context.Context, query string, args ...any) *sql
300281
return d.readDB.QueryRowContext(ctx, query, args...)
301282
}
302283

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.
284+
// Exec executes a statement via the write pool.
305285
func (d *database) Exec(ctx context.Context, query string, args ...any) (sql.Result, error) {
306-
result, err := d.db.ExecContext(ctx, query, args...)
307-
if err == nil {
308-
d.kickWatcher()
309-
}
310-
return result, err
286+
return d.db.ExecContext(ctx, query, args...)
311287
}
312288

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

‎internal/sqlite/sqlitetest/mock_db.go‎

Lines changed: 0 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -31,9 +31,6 @@ type MockDB struct {
3131
// LogEventPoliciesImpl is the mock log event policies implementation.
3232
LogEventPoliciesImpl sqlite.LogEventPolicies
3333

34-
// SubscriptionImpl is returned by Subscribe.
35-
SubscriptionImpl *sqlite.Subscription
36-
3734
// QueryFunc is called by Query.
3835
QueryFunc func(ctx context.Context, sql string, args ...any) (*sql.Rows, error)
3936

@@ -75,11 +72,6 @@ func (m *MockDB) ServiceStatuses() sqlite.ServiceStatuses {
7572
return m.ServiceStatusesImpl
7673
}
7774

78-
// Subscribe implements sqlite.DB.
79-
func (m *MockDB) Subscribe() *sqlite.Subscription {
80-
return m.SubscriptionImpl
81-
}
82-
8375
// Query implements sqlite.DB.
8476
func (m *MockDB) Query(ctx context.Context, sql string, args ...any) (*sql.Rows, error) {
8577
if m.QueryFunc != nil {

‎internal/sqlite/tables.go‎

Lines changed: 0 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -11,12 +11,3 @@ const (
1111
TableMessages Table = "messages"
1212
TableServices Table = "services"
1313
)
14-
15-
// knownTables maps PowerSync internal names to public Table constants.
16-
var knownTables = map[string]Table{
17-
"messages": TableMessages,
18-
"ps_data__messages": TableMessages,
19-
20-
"conversations": TableConversations,
21-
"ps_data__conversations": TableConversations,
22-
}

‎internal/sqlite/watch.go‎

Lines changed: 0 additions & 233 deletions
This file was deleted.

0 commit comments

Comments
 (0)