Monorepo for Tangled
appview/db: test state resolver, parking, reconcile queries
Lewis: May this revision serve well! <lewis@tangled.org>
+429
+429
appview/db/entity_state_test.go
+429
appview/db/entity_state_test.go
···
1
+
package db
2
+
3
+
import (
4
+
"testing"
5
+
6
+
"github.com/bluesky-social/indigo/atproto/syntax"
7
+
"tangled.org/core/appview/models"
8
+
"tangled.org/core/orm"
9
+
)
10
+
11
+
func seedIssue(t *testing.T, d *DB, repo *models.Repo, did, rkey string) *models.Issue {
12
+
t.Helper()
13
+
tx, err := d.Begin()
14
+
if err != nil {
15
+
t.Fatalf("Begin: %v", err)
16
+
}
17
+
issue := &models.Issue{
18
+
Did: did,
19
+
Rkey: rkey,
20
+
RepoDid: syntax.DID(repo.RepoDid),
21
+
Title: "title",
22
+
Body: "body",
23
+
Open: true,
24
+
}
25
+
if err := PutIssue(tx, issue); err != nil {
26
+
t.Fatalf("PutIssue: %v", err)
27
+
}
28
+
if err := tx.Commit(); err != nil {
29
+
t.Fatalf("Commit: %v", err)
30
+
}
31
+
return issue
32
+
}
33
+
34
+
func seedPull(t *testing.T, d *DB, repo *models.Repo, did, rkey string) *models.Pull {
35
+
t.Helper()
36
+
tx, err := d.Begin()
37
+
if err != nil {
38
+
t.Fatalf("Begin: %v", err)
39
+
}
40
+
pull := &models.Pull{
41
+
RepoDid: syntax.DID(repo.RepoDid),
42
+
OwnerDid: did,
43
+
Rkey: rkey,
44
+
Title: "title",
45
+
Body: "body",
46
+
TargetBranch: "main",
47
+
State: models.PullOpen,
48
+
}
49
+
if err := PutPull(tx, pull); err != nil {
50
+
t.Fatalf("PutPull: %v", err)
51
+
}
52
+
if err := tx.Commit(); err != nil {
53
+
t.Fatalf("Commit: %v", err)
54
+
}
55
+
return pull
56
+
}
57
+
58
+
func putIssueStateRec(t *testing.T, d *DB, rec models.StateRecord) {
59
+
t.Helper()
60
+
tx, err := d.Begin()
61
+
if err != nil {
62
+
t.Fatalf("Begin: %v", err)
63
+
}
64
+
if _, err := PutIssueState(tx, rec); err != nil {
65
+
t.Fatalf("PutIssueState: %v", err)
66
+
}
67
+
if err := ResolveIssueState(tx, rec.Subject); err != nil {
68
+
t.Fatalf("ResolveIssueState: %v", err)
69
+
}
70
+
if err := tx.Commit(); err != nil {
71
+
t.Fatalf("Commit: %v", err)
72
+
}
73
+
}
74
+
75
+
func deleteIssueStateRec(t *testing.T, d *DB, did, rkey string) {
76
+
t.Helper()
77
+
tx, err := d.Begin()
78
+
if err != nil {
79
+
t.Fatalf("Begin: %v", err)
80
+
}
81
+
subject, err := DeleteIssueState(tx, did, rkey)
82
+
if err != nil {
83
+
t.Fatalf("DeleteIssueState: %v", err)
84
+
}
85
+
if subject != "" {
86
+
if err := RecomputeIssueState(tx, subject); err != nil {
87
+
t.Fatalf("RecomputeIssueState: %v", err)
88
+
}
89
+
}
90
+
if err := tx.Commit(); err != nil {
91
+
t.Fatalf("Commit: %v", err)
92
+
}
93
+
}
94
+
95
+
func putPullStatusRec(t *testing.T, d *DB, rec models.StateRecord) {
96
+
t.Helper()
97
+
tx, err := d.Begin()
98
+
if err != nil {
99
+
t.Fatalf("Begin: %v", err)
100
+
}
101
+
if _, err := PutPullStatus(tx, rec); err != nil {
102
+
t.Fatalf("PutPullStatus: %v", err)
103
+
}
104
+
if err := ResolvePullStatus(tx, rec.Subject); err != nil {
105
+
t.Fatalf("ResolvePullStatus: %v", err)
106
+
}
107
+
if err := tx.Commit(); err != nil {
108
+
t.Fatalf("Commit: %v", err)
109
+
}
110
+
}
111
+
112
+
func issueOpen(t *testing.T, d *DB, subject syntax.ATURI) bool {
113
+
t.Helper()
114
+
issues, err := GetIssues(d, orm.FilterEq("at_uri", subject))
115
+
if err != nil || len(issues) != 1 {
116
+
t.Fatalf("GetIssues: %v len %d", err, len(issues))
117
+
}
118
+
return issues[0].Open
119
+
}
120
+
121
+
func pullStateOf(t *testing.T, d *DB, subject syntax.ATURI) models.PullState {
122
+
t.Helper()
123
+
pulls, err := GetPulls(d, orm.FilterEq("at_uri", subject))
124
+
if err != nil || len(pulls) != 1 {
125
+
t.Fatalf("GetPulls: %v len %d", err, len(pulls))
126
+
}
127
+
return pulls[0].State
128
+
}
129
+
130
+
func issueRec(did, rkey string, subject syntax.ATURI, v models.StateValue, micros int64) models.StateRecord {
131
+
return models.StateRecord{Did: did, Rkey: rkey, Subject: subject, Value: v, SortMicros: micros}
132
+
}
133
+
134
+
func TestIssueStateLastWriterWins(t *testing.T) {
135
+
d := newTestDB(t)
136
+
repo := seedRepo(t, d, "did:plc:akshay", "knot.example", "anemone", "anemone", "did:plc:anemone")
137
+
issue := seedIssue(t, d, repo, "did:plc:akshay", "issue1")
138
+
subject := issue.AtUri()
139
+
140
+
if !issueOpen(t, d, subject) {
141
+
t.Fatal("new issue should be open")
142
+
}
143
+
144
+
putIssueStateRec(t, d, issueRec("did:plc:akshay", "s2", subject, models.StateClosed, 200))
145
+
putIssueStateRec(t, d, issueRec("did:plc:akshay", "s1", subject, models.StateOpen, 100))
146
+
if issueOpen(t, d, subject) {
147
+
t.Fatal("earlier open@100 must not beat closed@200")
148
+
}
149
+
150
+
putIssueStateRec(t, d, issueRec("did:plc:akshay", "s3", subject, models.StateOpen, 300))
151
+
if !issueOpen(t, d, subject) {
152
+
t.Fatal("open@300 should win")
153
+
}
154
+
155
+
putIssueStateRec(t, d, issueRec("did:plc:akshay", "zzz", subject, models.StateClosed, 300))
156
+
if issueOpen(t, d, subject) {
157
+
t.Fatal("a tie at 300 must break to the greater source uri zzz=closed")
158
+
}
159
+
160
+
putIssueStateRec(t, d, issueRec("did:plc:akshay", "zzz", subject, models.StateClosed, 300))
161
+
if issueOpen(t, d, subject) {
162
+
t.Fatal("replaying the winning record must not change the result")
163
+
}
164
+
var count int
165
+
if err := d.QueryRow(`select count(*) from issue_states where subject = ?`, subject).Scan(&count); err != nil {
166
+
t.Fatalf("count: %v", err)
167
+
}
168
+
if count != 4 {
169
+
t.Fatalf("replaying an existing record must not duplicate rows, got %d want 4", count)
170
+
}
171
+
}
172
+
173
+
func TestIssueStateOrderIndependent(t *testing.T) {
174
+
build := func(order []int) bool {
175
+
d := newTestDB(t)
176
+
repo := seedRepo(t, d, "did:plc:akshay", "knot.example", "anemone", "anemone", "did:plc:anemone")
177
+
issue := seedIssue(t, d, repo, "did:plc:akshay", "issue1")
178
+
subject := issue.AtUri()
179
+
180
+
recs := []models.StateRecord{
181
+
issueRec("did:plc:akshay", "a", subject, models.StateOpen, 100),
182
+
issueRec("did:plc:akshay", "b", subject, models.StateClosed, 300),
183
+
issueRec("did:plc:akshay", "c", subject, models.StateOpen, 200),
184
+
}
185
+
for _, idx := range order {
186
+
putIssueStateRec(t, d, recs[idx])
187
+
}
188
+
return issueOpen(t, d, subject)
189
+
}
190
+
191
+
forward := build([]int{0, 1, 2})
192
+
shuffled := build([]int{2, 0, 1})
193
+
if forward != shuffled {
194
+
t.Fatalf("order changed result: forward=%v shuffled=%v", forward, shuffled)
195
+
}
196
+
if forward {
197
+
t.Fatal("highest-micros record closed@300 must win regardless of order")
198
+
}
199
+
}
200
+
201
+
func TestIssueStateDeleteRecomputes(t *testing.T) {
202
+
d := newTestDB(t)
203
+
repo := seedRepo(t, d, "did:plc:akshay", "knot.example", "anemone", "anemone", "did:plc:anemone")
204
+
issue := seedIssue(t, d, repo, "did:plc:akshay", "issue1")
205
+
subject := issue.AtUri()
206
+
207
+
putIssueStateRec(t, d, issueRec("did:plc:akshay", "s1", subject, models.StateClosed, 100))
208
+
putIssueStateRec(t, d, issueRec("did:plc:akshay", "s2", subject, models.StateOpen, 200))
209
+
if !issueOpen(t, d, subject) {
210
+
t.Fatal("open@200 should win before deletion")
211
+
}
212
+
213
+
deleteIssueStateRec(t, d, "did:plc:akshay", "s2")
214
+
if issueOpen(t, d, subject) {
215
+
t.Fatal("deleting open@200 must fall back to closed@100")
216
+
}
217
+
218
+
deleteIssueStateRec(t, d, "did:plc:akshay", "s1")
219
+
if !issueOpen(t, d, subject) {
220
+
t.Fatal("deleting the last state record must revert to open")
221
+
}
222
+
}
223
+
224
+
func TestIssueStateSubjectChangeRecomputesPrior(t *testing.T) {
225
+
d := newTestDB(t)
226
+
repo := seedRepo(t, d, "did:plc:akshay", "knot.example", "anemone", "anemone", "did:plc:anemone")
227
+
issueA := seedIssue(t, d, repo, "did:plc:akshay", "issueA")
228
+
issueB := seedIssue(t, d, repo, "did:plc:akshay", "issueB")
229
+
230
+
putIssueStateRec(t, d, issueRec("did:plc:akshay", "s1", issueA.AtUri(), models.StateClosed, 100))
231
+
if issueOpen(t, d, issueA.AtUri()) {
232
+
t.Fatal("issueA should be closed after closed@100")
233
+
}
234
+
235
+
tx, err := d.Begin()
236
+
if err != nil {
237
+
t.Fatalf("Begin: %v", err)
238
+
}
239
+
prior, err := PutIssueState(tx, issueRec("did:plc:akshay", "s1", issueB.AtUri(), models.StateClosed, 200))
240
+
if err != nil {
241
+
t.Fatalf("PutIssueState: %v", err)
242
+
}
243
+
if prior != issueA.AtUri() {
244
+
t.Fatalf("put must report the prior subject %s, got %q", issueA.AtUri(), prior)
245
+
}
246
+
if err := ResolveIssueState(tx, issueB.AtUri()); err != nil {
247
+
t.Fatalf("ResolveIssueState B: %v", err)
248
+
}
249
+
if err := RecomputeIssueState(tx, prior); err != nil {
250
+
t.Fatalf("RecomputeIssueState A: %v", err)
251
+
}
252
+
if err := tx.Commit(); err != nil {
253
+
t.Fatalf("Commit: %v", err)
254
+
}
255
+
256
+
if !issueOpen(t, d, issueA.AtUri()) {
257
+
t.Fatal("issueA must revert to open once its only state record repoints to issueB")
258
+
}
259
+
if issueOpen(t, d, issueB.AtUri()) {
260
+
t.Fatal("issueB should be closed after the record repoints to it")
261
+
}
262
+
}
263
+
264
+
func TestPullStatusLastWriterWins(t *testing.T) {
265
+
d := newTestDB(t)
266
+
repo := seedRepo(t, d, "did:plc:akshay", "knot.example", "limpet", "limpet", "did:plc:limpet")
267
+
pull := seedPull(t, d, repo, "did:plc:akshay", "pull1")
268
+
subject := pull.AtUri()
269
+
270
+
if pullStateOf(t, d, subject) != models.PullOpen {
271
+
t.Fatal("new pull should be open")
272
+
}
273
+
274
+
putPullStatusRec(t, d, issueRec("did:plc:akshay", "s1", subject, models.StateClosed, 100))
275
+
if pullStateOf(t, d, subject) != models.PullClosed {
276
+
t.Fatal("closed@100 should win")
277
+
}
278
+
279
+
putPullStatusRec(t, d, issueRec("did:plc:akshay", "s2", subject, models.StateMerged, 200))
280
+
if pullStateOf(t, d, subject) != models.PullMerged {
281
+
t.Fatal("merged@200 should win over closed@100")
282
+
}
283
+
284
+
putPullStatusRec(t, d, issueRec("did:plc:akshay", "s3", subject, models.StateOpen, 300))
285
+
if pullStateOf(t, d, subject) != models.PullOpen {
286
+
t.Fatal("open@300 should win over merged@200")
287
+
}
288
+
289
+
if err := AbandonPulls(d, orm.FilterEq("at_uri", subject)); err != nil {
290
+
t.Fatalf("AbandonPulls: %v", err)
291
+
}
292
+
putPullStatusRec(t, d, issueRec("did:plc:akshay", "s4", subject, models.StateOpen, 400))
293
+
if pullStateOf(t, d, subject) != models.PullAbandoned {
294
+
t.Fatal("an abandoned pull must not be resurrected by a later status record")
295
+
}
296
+
}
297
+
298
+
func TestIssueStateForeignKey(t *testing.T) {
299
+
d := newTestDB(t)
300
+
repo := seedRepo(t, d, "did:plc:akshay", "knot.example", "anemone", "anemone", "did:plc:anemone")
301
+
302
+
tx, err := d.Begin()
303
+
if err != nil {
304
+
t.Fatalf("Begin: %v", err)
305
+
}
306
+
ghost := syntax.ATURI("at://did:plc:akshay/sh.tangled.repo.issue/ghost")
307
+
if _, err := PutIssueState(tx, issueRec("did:plc:akshay", "s1", ghost, models.StateClosed, 100)); err == nil {
308
+
t.Fatal("inserting state for a nonexistent issue must violate the foreign key")
309
+
}
310
+
tx.Rollback()
311
+
312
+
issue := seedIssue(t, d, repo, "did:plc:akshay", "issue1")
313
+
subject := issue.AtUri()
314
+
putIssueStateRec(t, d, issueRec("did:plc:akshay", "s1", subject, models.StateClosed, 100))
315
+
316
+
dtx, err := d.Begin()
317
+
if err != nil {
318
+
t.Fatalf("Begin: %v", err)
319
+
}
320
+
if err := DeleteIssues(dtx, "did:plc:akshay", "issue1"); err != nil {
321
+
t.Fatalf("DeleteIssues: %v", err)
322
+
}
323
+
if err := dtx.Commit(); err != nil {
324
+
t.Fatalf("Commit: %v", err)
325
+
}
326
+
327
+
var remaining int
328
+
if err := d.QueryRow(`select count(*) from issue_states where subject = ?`, subject).Scan(&remaining); err != nil {
329
+
t.Fatalf("count: %v", err)
330
+
}
331
+
if remaining != 0 {
332
+
t.Fatalf("deleting the issue must cascade-delete its state rows, got %d", remaining)
333
+
}
334
+
}
335
+
336
+
func TestPendingStateRecords(t *testing.T) {
337
+
d := newTestDB(t)
338
+
s1 := syntax.ATURI("at://did:plc:boltless/sh.tangled.repo.issue/i1")
339
+
s2 := syntax.ATURI("at://did:plc:akshay/sh.tangled.repo.pull/p1")
340
+
issueNsid := "sh.tangled.repo.issue.state"
341
+
342
+
park := func(did, rkey, nsid string, subject syntax.ATURI, record string) {
343
+
t.Helper()
344
+
tx, err := d.Begin()
345
+
if err != nil {
346
+
t.Fatalf("Begin: %v", err)
347
+
}
348
+
if err := ParkStateRecord(tx, PendingStateRecord{
349
+
Did: did, Rkey: rkey, Nsid: nsid, Subject: subject, Record: []byte(record),
350
+
}); err != nil {
351
+
t.Fatalf("ParkStateRecord: %v", err)
352
+
}
353
+
if err := tx.Commit(); err != nil {
354
+
t.Fatalf("Commit: %v", err)
355
+
}
356
+
}
357
+
358
+
park("did:plc:boltless", "s1", issueNsid, s1, `{"v":1}`)
359
+
park("did:plc:boltless", "s1", issueNsid, s1, `{"v":2}`)
360
+
park("did:plc:akshay", "p1", "sh.tangled.repo.pull.status", s2, `{}`)
361
+
362
+
pending, err := PendingStateRecordsForSubject(d, s1)
363
+
if err != nil {
364
+
t.Fatalf("query: %v", err)
365
+
}
366
+
if len(pending) != 1 || pending[0].Did != "did:plc:boltless" || string(pending[0].Record) != `{"v":2}` {
367
+
t.Fatalf("re-park must overwrite without duplicating, got %+v", pending)
368
+
}
369
+
370
+
subjects, err := DistinctPendingStateSubjects(d)
371
+
if err != nil {
372
+
t.Fatalf("DistinctPendingStateSubjects: %v", err)
373
+
}
374
+
if len(subjects) != 2 {
375
+
t.Fatalf("want 2 distinct subjects from 3 parked rows, got %d", len(subjects))
376
+
}
377
+
378
+
tx, err := d.Begin()
379
+
if err != nil {
380
+
t.Fatalf("Begin: %v", err)
381
+
}
382
+
if err := UnparkStateRecord(tx, "did:plc:boltless", "s1", issueNsid); err != nil {
383
+
t.Fatalf("UnparkStateRecord: %v", err)
384
+
}
385
+
if err := tx.Commit(); err != nil {
386
+
t.Fatalf("Commit: %v", err)
387
+
}
388
+
if pending, _ := PendingStateRecordsForSubject(d, s1); len(pending) != 0 {
389
+
t.Fatalf("want 0 after unpark, got %d", len(pending))
390
+
}
391
+
}
392
+
393
+
func TestEvictStalePendingStateRecords(t *testing.T) {
394
+
d := newTestDB(t)
395
+
repo := seedRepo(t, d, "did:plc:akshay", "knot.example", "anemone", "anemone", "did:plc:anemone")
396
+
issue := seedIssue(t, d, repo, "did:plc:akshay", "issue1")
397
+
live := issue.AtUri()
398
+
orphan := syntax.ATURI("at://did:plc:boltless/sh.tangled.repo.issue/ghost")
399
+
400
+
insert := func(rkey, created string, subject syntax.ATURI) {
401
+
t.Helper()
402
+
if _, err := d.Exec(
403
+
`insert into pending_state_records (did, rkey, nsid, subject, record, created) values (?, ?, ?, ?, ?, ?)`,
404
+
"did:plc:boltless", rkey, "sh.tangled.repo.issue.state", string(subject), []byte("{}"), created,
405
+
); err != nil {
406
+
t.Fatalf("insert %s: %v", rkey, err)
407
+
}
408
+
}
409
+
410
+
insert("stale-orphan", "2000-01-01T00:00:00Z", orphan)
411
+
insert("fresh-orphan", "2999-01-01T00:00:00Z", orphan)
412
+
insert("stale-live", "2000-01-01T00:00:00Z", live)
413
+
414
+
evicted, err := EvictStalePendingStateRecords(d, "2026-01-01T00:00:00Z")
415
+
if err != nil {
416
+
t.Fatalf("EvictStalePendingStateRecords: %v", err)
417
+
}
418
+
if evicted != 1 {
419
+
t.Fatalf("only the stale orphan must be evicted, got %d want 1", evicted)
420
+
}
421
+
422
+
var remaining int
423
+
if err := d.QueryRow(`select count(*) from pending_state_records`).Scan(&remaining); err != nil {
424
+
t.Fatalf("count: %v", err)
425
+
}
426
+
if remaining != 2 {
427
+
t.Fatalf("the fresh orphan and the stale row with a live subject must survive, got %d want 2", remaining)
428
+
}
429
+
}