oficina / slides Public

Download this file

1	package slides
2
3	import (
4	"context"
5	"fmt"
6	"gorm.io/gorm"
7	"strings"
8	"testing"
9	"time"
10
11	"amadan.net/oficina/slides/internal/slides/core"
12	)
13
14	func TestAuthoritativeLogBeyondCoreRetentionLimit(t *testing.T) {
15	db := testDB(t)
16	a, _, err := newApp(db, Config{Env: EnvDevelopment, Logger: quietLogger()})
17	if err != nil {
18	t.Fatal(err)
19	}
20	t.Cleanup(func() { a.Close() })
21	ctx := context.Background()
22	owner, _, err := ensureUser(ctx, db.G, "long-log@example.test")
23	if err != nil {
24	t.Fatal(err)
25	}
26	row, err := a.createDeckRow(ctx, owner, "Long history", defaultTheme)
27	if err != nil {
28	t.Fatal(err)
29	}
30	h, err := a.core.NewDoc(ctx, row.Sqid, serverClientID)
31	if err != nil {
32	t.Fatal(err)
33	}
34	defer func() { h.Close(ctx) }()
35	updates := make([]DeckUpdate, 0, core.MaxRetainedUpdates+19)
36	for seq := int64(1); seq <= core.MaxRetainedUpdates+19; seq++ {
37	if seq%1000 == 0 {
38	state, err := h.State(ctx)
39	if err != nil {
40	t.Fatal(err)
41	}
42	h.Close(ctx)
43	h, err = a.core.NewDoc(ctx, row.Sqid, serverClientID)
44	if err != nil {
45	t.Fatal(err)
46	}
47	if err := h.Apply(ctx, state); err != nil {
48	t.Fatal(err)
49	}
50	}
51	batch, err := opsJSON([]op{{"op": "set_meta", "title": fmt.Sprintf("Revision %d", seq)}})
52	if err != nil {
53	t.Fatal(err)
54	}
55	if seq == 1 {
56	batch, err = openingOps("Long history", defaultTheme)
57	if err != nil {
58	t.Fatal(err)
59	}
60	}
61	update, err := h.Ops(ctx, batch)
62	if err != nil {
63	t.Fatal(err)
64	}
65	updates = append(updates, DeckUpdate{DeckID: row.ID, Seq: seq, Actor: actorFor(owner), Bytes: update, CreatedAt: time.Now().UTC()})
66	row.Seq = seq
67	}
68	if err := db.G.CreateInBatches(updates, 128).Error; err != nil {
69	t.Fatal(err)
70	}
71	if err := db.G.Model(&Deck{}).Where("id = ?", row.ID).Update("seq", row.Seq).Error; err != nil {
72	t.Fatal(err)
73	}
74	want, err := h.JSON(ctx)
75	if err != nil {
76	t.Fatal(err)
77	}
78	got := viewJSON(t, a, row)
79	if string(got) != string(want) {
80	t.Fatal("long replay changed document content")
81	}
82	t.Run("bounded_queries", func(t *testing.T) {
83	queries := 0
84	name := "test:bounded_replay"
85	if err := db.G.Callback().Query().After("gorm:query").Register(name, func(tx *gorm.DB) {
86	if tx.Statement.Table == "deck_updates" {
87	queries++
88	if tx.RowsAffected > 256 {
89	t.Errorf("unbounded database batch: %d rows", tx.RowsAffected)
90	}
91	}
92	}); err != nil {
93	t.Fatal(err)
94	}
95	defer db.G.Callback().Query().Remove(name)
96	if string(viewJSON(t, a, row)) != string(want) {
97	t.Fatal("bounded query changed content")
98	}
99	if queries < 2 {
100	t.Fatal("long log was not paged")
101	}
102	})
103	t.Run("unreadable_cache", func(t *testing.T) {
104	snap := DeckSnapshot{DeckID: row.ID, Seq: 128, Bytes: []byte{0xff}, CreatedAt: time.Now().UTC()}
105	if err := db.G.Create(&snap).Error; err != nil {
106	t.Fatal(err)
107	}
108	defer db.G.Delete(&snap)
109	if string(viewJSON(t, a, row)) != string(want) {
110	t.Fatal("corrupt cache changed long replay")
111	}
112	})
113	t.Run("future_cache_and_update", func(t *testing.T) {
114	batch, err := opsJSON([]op{{"op": "set_meta", "title": "Future revision"}})
115	if err != nil {
116	t.Fatal(err)
117	}
118	update, err := h.Ops(ctx, batch)
119	if err != nil {
120	t.Fatal(err)
121	}
122	state, err := h.State(ctx)
123	if err != nil {
124	t.Fatal(err)
125	}
126	future := DeckUpdate{DeckID: row.ID, Seq: row.Seq + 1, Actor: actorFor(owner), Bytes: update, CreatedAt: time.Now().UTC()}
127	if err := db.G.Create(&future).Error; err != nil {
128	t.Fatal(err)
129	}
130	defer db.G.Delete(&future)
131	snap := DeckSnapshot{DeckID: row.ID, Seq: row.Seq + 1, Bytes: state, CreatedAt: time.Now().UTC()}
132	if err := db.G.Create(&snap).Error; err != nil {
133	t.Fatal(err)
134	}
135	defer db.G.Delete(&snap)
136	if string(viewJSON(t, a, row)) != string(want) {
137	t.Fatal("replay included a revision after its captured coordinate")
138	}
139	})
140	for _, missing := range []int64{256, row.Seq} {
141	t.Run(fmt.Sprintf("missing_%d", missing), func(t *testing.T) {
142	u := updates[missing-1]
143	if err := db.G.Delete(&u).Error; err != nil {
144	t.Fatal(err)
145	}
146	defer func() {
147	if err := db.G.Create(&u).Error; err != nil {
148	t.Error(err)
149	}
150	}()
151	replayed, view, err := a.openDeck(ctx, row)
152	if replayed != nil {
153	replayed.Close(ctx)
154	}
155	if err == nil || replayed != nil || view != nil {
156	t.Fatal("incomplete log returned a partial deck")
157	}
158	})
159	}
160	t.Run("canceled_checkpoint", func(t *testing.T) {
161	cancelCtx, cancel := context.WithCancel(ctx)
162	defer cancel()
163	queries := 0
164	name := "test:cancel_checkpoint"
165	if err := db.G.Callback().Query().After("gorm:query").Register(name, func(tx *gorm.DB) {
166	if tx.Statement.Table == "deck_updates" {
167	queries++
168	if queries == 9 {
169	cancel()
170	}
171	}
172	}); err != nil {
173	t.Fatal(err)
174	}
175	defer db.G.Callback().Query().Remove(name)
176	replayed, view, err := a.openDeck(cancelCtx, row)
177	if replayed != nil {
178	replayed.Close(ctx)
179	}
180	if err == nil || replayed != nil || view != nil {
181	t.Fatal("canceled replay returned a partial deck")
182	}
183	if !strings.Contains(err.Error(), "checkpoint") {
184	t.Fatalf("did not reach the canceled State call: %v", err)
185	}
186	})
187	opened, _, err := a.openDeck(ctx, row)
188	if err != nil {
189	t.Fatal(err)
190	}
191	defer opened.Close(ctx)
192	batch, err := opsJSON([]op{{"op": "set_meta", "title": "After replay"}})
193	if err != nil {
194	t.Fatal(err)
195	}
196	if _, err := opened.Ops(ctx, batch); err != nil {
197	t.Fatalf("replayed handle cannot accept next edit: %v", err)
198	}
199	var count int64
200	if err := db.G.Model(&DeckUpdate{}).Where("deck_id = ?", row.ID).Count(&count).Error; err != nil {
201	t.Fatal(err)
202	}
203	if count != row.Seq {
204	t.Fatal("read rewrote authoritative history")
205	}
206	if err := db.G.Model(&DeckSnapshot{}).Where("deck_id = ?", row.ID).Count(&count).Error; err != nil {
207	t.Fatal(err)
208	}
209	if count != 0 {
210	t.Fatal("read persisted an accelerator snapshot")
211	}
212	}
213