Monorepo for Tangled
tangled.org
knotserver/db: add owner UID assignment and repo isolation
author
Anirudh Oppiliappan
committer
Tangled
date
(Jun 12, 2026, 12:22 PM +0300)
commit
eb83c202
eb83c202bad5b920ed4612102369f7d28b0a23fb
parent
390ef2a1
390ef2a1d90784599eff69f0d6dec919887417a9
change-id
omlykowv
omlykowvspvmlnzzoxrutpnqrzuqqnyv
+237
-7
+44
-7
knotserver/db/db.go
+44
-7
knotserver/db/db.go
···
7
7
"log/slog"
8
8
"os"
9
9
"strings"
10
+
"sync"
10
11
11
12
securejoin "github.com/cyphar/filepath-securejoin"
12
13
_ "github.com/mattn/go-sqlite3"
···
17
18
type DB struct {
18
19
db *sql.DB
19
20
logger *slog.Logger
21
+
22
+
// uidAssignMu serialises GetOrAssignOwnerUID across goroutines so that
23
+
// concurrent callers don't race on the uid_counter read-modify-write.
24
+
uidAssignMu sync.Mutex
20
25
}
21
26
22
27
type DBTX interface {
···
193
198
194
199
if err := orm.RunMigration(conn, logger, "create-knot-members", func(tx *sql.Tx) error {
195
200
_, mErr := tx.ExecContext(ctx, `
196
-
create table if not exists knot_members (
197
-
id integer primary key autoincrement,
198
-
did text not null,
199
-
rkey text not null,
200
-
subject text not null,
201
-
created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
202
-
unique (did, rkey)
201
+
create table if not exists knot_members (
202
+
id integer primary key autoincrement,
203
+
did text not null,
204
+
rkey text not null,
205
+
subject text not null,
206
+
created text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
207
+
unique (did, rkey)
208
+
);
209
+
`)
210
+
return mErr
211
+
}); err != nil {
212
+
return nil, err
213
+
}
214
+
215
+
if err := orm.RunMigration(conn, logger, "add-isolated-at-to-repo-keys", func(tx *sql.Tx) error {
216
+
_, mErr := tx.ExecContext(ctx, `ALTER TABLE repo_keys ADD COLUMN isolated_at DATETIME`)
217
+
return mErr
218
+
}); err != nil {
219
+
return nil, err
220
+
}
221
+
222
+
if err := orm.RunMigration(conn, logger, "add-owner-uid-tables", func(tx *sql.Tx) error {
223
+
_, mErr := tx.ExecContext(ctx, `
224
+
CREATE TABLE IF NOT EXISTS owner_uid_assignments (
225
+
owner_did TEXT PRIMARY KEY,
226
+
uid INTEGER NOT NULL UNIQUE,
227
+
created_at DATETIME DEFAULT CURRENT_TIMESTAMP
203
228
);
229
+
230
+
CREATE TABLE IF NOT EXISTS uid_counter (
231
+
next_uid INTEGER NOT NULL DEFAULT 100000
232
+
);
233
+
`)
234
+
if mErr != nil {
235
+
return mErr
236
+
}
237
+
// Seed the counter only if the table is empty.
238
+
_, mErr = tx.ExecContext(ctx, `
239
+
INSERT INTO uid_counter (next_uid)
240
+
SELECT 100000 WHERE NOT EXISTS (SELECT 1 FROM uid_counter)
204
241
`)
205
242
return mErr
206
243
}); err != nil {
+102
knotserver/db/owner_uid.go
+102
knotserver/db/owner_uid.go
···
1
+
package db
2
+
3
+
import (
4
+
"database/sql"
5
+
)
6
+
7
+
// GetOrAssignOwnerUID returns the virtual UID for ownerDID, minting a new one
8
+
// from the uid_counter table if this owner has not been seen before.
9
+
// UIDs start at 100000 and increment by one per unique owner.
10
+
//
11
+
// A process-wide mutex serialises concurrent callers so two simultaneous
12
+
// requests for distinct DIDs do not race to claim the same counter value.
13
+
// The mutex is local to this DB instance and SQLite itself only permits one
14
+
// writer at a time, so this does not impede throughput in practice.
15
+
func (d *DB) GetOrAssignOwnerUID(ownerDID string) (uint32, error) {
16
+
d.uidAssignMu.Lock()
17
+
defer d.uidAssignMu.Unlock()
18
+
19
+
tx, err := d.db.Begin()
20
+
if err != nil {
21
+
return 0, err
22
+
}
23
+
defer tx.Rollback()
24
+
25
+
var uid uint32
26
+
err = tx.QueryRow(
27
+
`SELECT uid FROM owner_uid_assignments WHERE owner_did = ?`,
28
+
ownerDID,
29
+
).Scan(&uid)
30
+
if err == nil {
31
+
return uid, tx.Commit()
32
+
}
33
+
if err != sql.ErrNoRows {
34
+
return 0, err
35
+
}
36
+
37
+
if err := tx.QueryRow(`SELECT next_uid FROM uid_counter`).Scan(&uid); err != nil {
38
+
return 0, err
39
+
}
40
+
if _, err := tx.Exec(`UPDATE uid_counter SET next_uid = next_uid + 1`); err != nil {
41
+
return 0, err
42
+
}
43
+
if _, err := tx.Exec(
44
+
`INSERT INTO owner_uid_assignments (owner_did, uid) VALUES (?, ?)`,
45
+
ownerDID, uid,
46
+
); err != nil {
47
+
return 0, err
48
+
}
49
+
50
+
return uid, tx.Commit()
51
+
}
52
+
53
+
// AllReposForMigration returns all (repo_did, owner_did) pairs with a
54
+
// non-null owner. Pass force=true to include already-migrated repos.
55
+
func (d *DB) AllReposForMigration(force bool) ([]RepoMigrationRow, error) {
56
+
query := `SELECT repo_did, owner_did FROM repo_keys WHERE owner_did IS NOT NULL`
57
+
if !force {
58
+
query += ` AND isolated_at IS NULL`
59
+
}
60
+
rows, err := d.db.Query(query)
61
+
if err != nil {
62
+
return nil, err
63
+
}
64
+
defer rows.Close()
65
+
66
+
var result []RepoMigrationRow
67
+
for rows.Next() {
68
+
var r RepoMigrationRow
69
+
if err := rows.Scan(&r.RepoDID, &r.OwnerDID); err != nil {
70
+
return nil, err
71
+
}
72
+
result = append(result, r)
73
+
}
74
+
return result, rows.Err()
75
+
}
76
+
77
+
// CountUnmigratedRepos returns the number of repos that have not yet been
78
+
// isolation-migrated.
79
+
func (d *DB) CountUnmigratedRepos() (int, error) {
80
+
var n int
81
+
err := d.db.QueryRow(`
82
+
SELECT count(1) FROM repo_keys
83
+
WHERE owner_did IS NOT NULL
84
+
AND isolated_at IS NULL
85
+
`).Scan(&n)
86
+
return n, err
87
+
}
88
+
89
+
// MarkRepoIsolated sets isolated_at to the current time for repoDID.
90
+
func (d *DB) MarkRepoIsolated(repoDID string) error {
91
+
_, err := d.db.Exec(
92
+
`UPDATE repo_keys SET isolated_at = strftime('%Y-%m-%dT%H:%M:%SZ','now') WHERE repo_did = ?`,
93
+
repoDID,
94
+
)
95
+
return err
96
+
}
97
+
98
+
// RepoMigrationRow is a row returned by AllReposForMigration.
99
+
type RepoMigrationRow struct {
100
+
RepoDID string
101
+
OwnerDID string
102
+
}
+91
knotserver/db/owner_uid_test.go
+91
knotserver/db/owner_uid_test.go
···
1
+
package db
2
+
3
+
import (
4
+
"context"
5
+
"fmt"
6
+
"path/filepath"
7
+
"sync"
8
+
"testing"
9
+
)
10
+
11
+
// TestGetOrAssignOwnerUID_Concurrent verifies that concurrent calls for
12
+
// distinct DIDs each get a unique UID, with no duplicates from a race on
13
+
// the uid_counter table.
14
+
func TestGetOrAssignOwnerUID_Concurrent(t *testing.T) {
15
+
d := newTestDB(t)
16
+
17
+
const n = 50
18
+
results := make([]uint32, n)
19
+
var wg sync.WaitGroup
20
+
wg.Add(n)
21
+
for i := 0; i < n; i++ {
22
+
go func(i int) {
23
+
defer wg.Done()
24
+
uid, err := d.GetOrAssignOwnerUID(fmt.Sprintf("did:plc:test-%03d", i))
25
+
if err != nil {
26
+
t.Errorf("GetOrAssignOwnerUID(%d): %v", i, err)
27
+
return
28
+
}
29
+
results[i] = uid
30
+
}(i)
31
+
}
32
+
wg.Wait()
33
+
34
+
seen := map[uint32]int{}
35
+
for i, uid := range results {
36
+
if uid == 0 {
37
+
t.Errorf("result[%d] is zero (probably an error above)", i)
38
+
continue
39
+
}
40
+
if prev, ok := seen[uid]; ok {
41
+
t.Errorf("uid %d was returned for both index %d and %d", uid, prev, i)
42
+
}
43
+
seen[uid] = i
44
+
}
45
+
46
+
if len(seen) != n {
47
+
t.Errorf("got %d unique UIDs, want %d", len(seen), n)
48
+
}
49
+
}
50
+
51
+
// TestGetOrAssignOwnerUID_Idempotent verifies that repeated calls for the
52
+
// same DID return the same UID.
53
+
func TestGetOrAssignOwnerUID_Idempotent(t *testing.T) {
54
+
d := newTestDB(t)
55
+
const did = "did:plc:stable"
56
+
57
+
first, err := d.GetOrAssignOwnerUID(did)
58
+
if err != nil {
59
+
t.Fatalf("first call: %v", err)
60
+
}
61
+
for i := 0; i < 5; i++ {
62
+
uid, err := d.GetOrAssignOwnerUID(did)
63
+
if err != nil {
64
+
t.Fatalf("repeat call %d: %v", i, err)
65
+
}
66
+
if uid != first {
67
+
t.Errorf("repeat call %d returned %d, want %d", i, uid, first)
68
+
}
69
+
}
70
+
}
71
+
72
+
// TestGetOrAssignOwnerUID_StartsAt100000 verifies the counter seeds correctly.
73
+
func TestGetOrAssignOwnerUID_StartsAt100000(t *testing.T) {
74
+
d := newTestDB(t)
75
+
uid, err := d.GetOrAssignOwnerUID("did:plc:first")
76
+
if err != nil {
77
+
t.Fatalf("first call: %v", err)
78
+
}
79
+
if uid != 100000 {
80
+
t.Errorf("first UID = %d, want 100000", uid)
81
+
}
82
+
}
83
+
84
+
func newTestDB(t *testing.T) *DB {
85
+
t.Helper()
86
+
d, err := Setup(context.Background(), filepath.Join(t.TempDir(), "test.db"))
87
+
if err != nil {
88
+
t.Fatalf("db.Setup: %v", err)
89
+
}
90
+
return d
91
+
}