carlos / platform Public

Download this file

1	package registry
2
3	import (
4	"errors"
5	"fmt"
6	"path/filepath"
7	"testing"
8	"time"
9	)
10
11	func TestClaimReviewOpenRetainsNormalDurability(t *testing.T) {
12	t.Setenv("CARLOS_HOST_CLAIM", "")
13	r, err := Open(filepath.Join(t.TempDir(), "registry.db"))
14	if err != nil {
15	t.Fatal(err)
16	}
17	defer r.Close()
18	var mode int
19	if err := r.sql.QueryRow("PRAGMA synchronous").Scan(&mode); err != nil {
20	t.Fatal(err)
21	}
22	if mode != 1 {
23	t.Fatalf("flag-off synchronous=%d, want NORMAL (1)", mode)
24	}
25	}
26
27	func TestClaimReviewFlagOffTimersKeepLegacyStatements(t *testing.T) {
28	t.Setenv("CARLOS_HOST_CLAIM", "")
29	r, err := Open(filepath.Join(t.TempDir(), "registry.db"))
30	if err != nil {
31	t.Fatal(err)
32	}
33	defer r.Close()
34	if _, err := r.sql.Exec("DROP TABLE routes"); err != nil {
35	t.Fatal(err)
36	}
37	timer := Timer{Host: "app.test", Name: "tick", Kind: "schedule", Spec: "every 1h0m0s", Path: "/tick", NextAt: 12, DueAt: 12}
38	if err := r.UpsertTimer(timer); err != nil {
39	t.Errorf("legacy upsert unexpectedly reads routes: %v", err)
40	}
41	timer.Registration = "imported"
42	if err := r.ReplaceTimers(timer.Host, []Timer{timer}); err != nil {
43	t.Fatalf("legacy replace unexpectedly reads routes: %v", err)
44	}
45	got, err := r.GetTimer(timer.Host, timer.Name)
46	if err != nil || *got != timer {
47	t.Fatalf("legacy import=%+v err=%v", got, err)
48	}
49	}
50
51	func TestClaimReviewTimerWritersWaitForOtherConnection(t *testing.T) {
52	t.Setenv("CARLOS_HOST_CLAIM", "1")
53	path := filepath.Join(t.TempDir(), "registry.db")
54	r, err := Open(path)
55	if err != nil {
56	t.Fatal(err)
57	}
58	defer r.Close()
59	other, err := Open(path)
60	if err != nil {
61	t.Fatal(err)
62	}
63	defer other.Close()
64	timer := Timer{Host: "app.test", Name: "tick", Kind: "schedule", Path: "/tick", Registration: "imported"}
65	for _, replace := range []bool{false, true} {
66	for i := 0; i < 100; i++ {
67	tx, err := other.sql.Begin()
68	if err != nil {
69	t.Fatal(err)
70	}
71	if _, err := tx.Exec("UPDATE routes SET status=status"); err != nil {
72	t.Fatal(err)
73	}
74	done := make(chan error, 1)
75	go func() {
76	if replace {
77	done <- r.ReplaceTimers(timer.Host, []Timer{timer})
78	} else {
79	done <- r.UpsertTimer(timer)
80	}
81	}()
82	time.Sleep(time.Millisecond)
83	if err := tx.Commit(); err != nil {
84	t.Fatal(err)
85	}
86	if err := <-done; err != nil {
87	t.Fatalf("replace=%v writer failed instead of waiting: %v", replace, err)
88	}
89	}
90	}
91	}
92
93	func TestClaimReviewOwnershipMigrationDoesNotOwnHoldColumn(t *testing.T) {
94	r, err := Open(filepath.Join(t.TempDir(), "registry.db"))
95	if err != nil {
96	t.Fatal(err)
97	}
98	defer r.Close()
99	if _, err := r.sql.Exec("ALTER TABLE routes DROP COLUMN hold_op"); err != nil {
100	t.Fatal(err)
101	}
102	if err := r.migrateClaimOwnership(); err != nil {
103	t.Fatal(err)
104	}
105	if _, err := r.sql.Exec("SELECT hold_op FROM routes"); err == nil {
106	t.Fatal("claim ownership migration redundantly added main's hold_op")
107	}
108	}
109
110	func TestClaimReviewDurabilityScopedToClaimWrites(t *testing.T) {
111	r, err := Open(filepath.Join(t.TempDir(), "registry.db"))
112	if err != nil {
113	t.Fatal(err)
114	}
115	defer r.Close()
116	rt := Route{Host: "app.test"}
117	if err := r.Upsert(rt); err != nil {
118	t.Fatal(err)
119	}
120	timer := Timer{Host: rt.Host, Name: "tick", Kind: "schedule"}
121	if err := r.UpsertTimer(timer); err != nil {
122	t.Fatal(err)
123	}
124	observed, err := r.Get(rt.Host)
125	if err != nil {
126	t.Fatal(err)
127	}
128	observedTimer, err := r.GetTimer(rt.Host, timer.Name)
129	if err != nil {
130	t.Fatal(err)
131	}
132	for _, table := range []string{"routes", "timers", "claim_row_retirements"} {
133	for _, op := range []string{"UPDATE", "DELETE", "INSERT"} {
134	q := fmt.Sprintf(`CREATE TRIGGER require_full_%s_%s BEFORE %s ON %s BEGIN SELECT CASE WHEN (SELECT synchronous FROM pragma_synchronous) != 2 THEN RAISE(ABORT,'claim write without FULL durability') END; END`, table, op, op, table)
135	if _, err := r.sql.Exec(q); err != nil {
136	t.Fatal(err)
137	}
138	}
139	}
140	check := func(err error) {
141	t.Helper()
142	if err != nil {
143	t.Fatal(err)
144	}
145	var mode int
146	if err := r.sql.QueryRow("PRAGMA synchronous").Scan(&mode); err != nil || mode != 1 {
147	t.Fatalf("normal durability not restored: %d %v", mode, err)
148	}
149	}
150	owner := ClaimOwnership{"inc", 1, "owner", "route"}
151	check(r.AdoptClaimRoute(t.Context(), *observed, owner))
152	timerOwner := owner
153	timerOwner.ClaimRowID = observedTimer.Registration
154	check(r.AdoptClaimTimer(t.Context(), *observedTimer, timerOwner))
155	check(r.SetClaimHold(t.Context(), rt.Host, owner, "", "hold"))
156	if err := r.SetClaimHold(t.Context(), rt.Host, owner, "wrong", ""); err != ErrClaimOwnership {
157	t.Fatal(err)
158	}
159	check(nil)
160	targets := []ClaimRow{{Kind: "route", Host: rt.Host, Ownership: owner}, {Kind: "timer", Host: rt.Host, Name: timer.Name, Registration: observedTimer.Registration, Ownership: timerOwner}}
161	failed := errors.New("archive failed")
162	if err := r.RetireClaimRows(t.Context(), "retire", targets, func([]ClaimRowArchive) error { return failed }); !errors.Is(err, failed) {
163	t.Fatal(err)
164	}
165	check(nil)
166	check(r.RetireClaimRows(t.Context(), "retire", targets, func([]ClaimRowArchive) error { return nil }))
167	}
168