~james/core

core/spindle/collaborator_marks_test.go -rw-r--r-- 3.1 KiB
452f17b7 — james chore(import): snapshot james.dad.town/core at f70c71c7696a 16 hours ago
                                                                                
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
package spindle

import (
	"context"
	"fmt"
	"net/http"
	"sync"
	"testing"

	"github.com/bluesky-social/indigo/atproto/syntax"

	"tangled.org/core/api/org_tangled"
	"tangled.org/core/jetstream"
)

func (e *ensureEnv) markCollaboratorEvent(t *testing.T, who syntax.DID, op string) {
	t.Helper()
	evt, ok := jetstreamToTapEvent(collaboratorCommit(testRepoDid, who, op))
	if !ok {
		t.Fatalf("no tap event for %s", op)
	}
	if err := e.s.processEvent(withMarkOnly(context.Background()), evt); err != nil {
		t.Fatal(err)
	}
}

func TestAPastTheCapRepoStillTakesAGrantAndARevokeOneRecordAtATime(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo", testBob)
	env.ensure(t)
	env.wantPush(t, testBob, true)

	// 3300 is one page past what the knot client reads, which means a sync can't finish in one go
	for i := range 3300 {
		env.grant(syntax.DID(fmt.Sprintf("did:plc:bulk%d", i)))
	}

	env.revoke(testBob)
	env.markCollaboratorEvent(t, testBob, jetstream.CommitOperationDelete)
	env.grant(testCarol)
	env.markCollaboratorEvent(t, testCarol, jetstream.CommitOperationCreate)

	before := env.knot.count()
	env.s.runRepoJob(context.Background(), testRepoDid)

	env.wantNoPushFor(t, testBob)
	env.wantPush(t, testCarol, true)
	if got := env.knot.count() - before; got > 40 {
		t.Fatalf("the job made %d knot requests, one failed sync is about 35 and each record is one more", got)
	}
	if got := dueRecords(t, env); len(got) != 0 {
		t.Fatalf("collaborator marks left behind: %+v", got)
	}
	if _, marked, err := env.s.db.RepoRetryRow(testRepoDid); err != nil || marked {
		t.Fatalf("a repo that can't be synced whole keeps retrying, exists %v, %v", marked, err)
	}
}

func TestACollaboratorEventOnASyncableRepoIsCoveredByTheFullSync(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo", testBob)
	env.ensure(t)

	env.grant(testCarol)
	env.markCollaboratorEvent(t, testCarol, jetstream.CommitOperationCreate)
	before := env.knot.count()
	env.s.runRepoJob(context.Background(), testRepoDid)

	env.wantPush(t, testCarol, true)
	if got := env.knot.count() - before; got > 3 {
		t.Fatalf("the job made %d knot requests, a refresh is 3 and the collaborator record needs no read of its own", got)
	}
	if got := dueRecords(t, env); len(got) != 0 {
		t.Fatalf("the full sync left collaborator marks: %+v", got)
	}
}

func TestACollaboratorMarkAppliedAfterAWipeDoesNotGrantAccessToTheWipedRepo(t *testing.T) {
	env := newEnsureEnv(t)
	env.serve("repo")
	env.ensure(t)
	env.grant(testBob)
	env.markCollaboratorEvent(t, testBob, jetstream.CommitOperationCreate)

	// the wipe lands during the mark's knot lookup, fired by the first collaborator query
	var wiped sync.Once
	env.knot.beforeAnswer = func(r *http.Request) {
		if r.URL.Query().Get("collection") != org_tangled.GraphCollaboratorNSID {
			return
		}
		wiped.Do(func() {
			if err := env.s.WipeRepo(context.Background(), testRepoDid, "test"); err != nil {
				t.Errorf("WipeRepo: %v", err)
			}
		})
	}
	env.s.runRecordMarks(context.Background(), testRepoDid)

	if rows, err := env.s.db.ListCollaboratorsByRepoDid(testRepoDid); err != nil || len(rows) != 0 {
		t.Fatalf("collaborator rows after the wipe = %+v, %v, want none", rows, err)
	}
	env.wantNoPushFor(t, testBob)
}