tinyrelayfile
tinyrelay/internal/daemon/push_followups_test.go/
internal/daemon/push_followups_test.go raw
1package daemon
2
3import (
4 "context"
5 "testing"
6 "time"
7
8 "github.com/FelineStateMachine/tinyrelay/internal/event"
9 "github.com/FelineStateMachine/tinyrelay/internal/relay"
10 "github.com/FelineStateMachine/tinyrelay/internal/storage"
11)
12
13func enablePushPlanning(t *testing.T, tenant *Tenant) {
14 t.Helper()
15 tenant.app.cfg.PushCallbackOrigins = []string{"https://push.example"}
16 p := tenant.Policy()
17 p.Features.Push = true
18 p.PushCallbacks = []string{"https://push.example"}
19 if err := tenant.applyPolicy(context.Background(), p); err != nil {
20 t.Fatal(err)
21 }
22}
23
24func makeReplicationPushRegistration(t *testing.T, tenant *Tenant, d string, created int64, kinds ...int) event.Event {
25 t.Helper()
26 if len(kinds) == 0 {
27 kinds = []int{1}
28 }
29 filter := `{"kinds":[1]}`
30 if kinds[0] != 1 {
31 filter = `{"kinds":[30618]}`
32 }
33 registration := event.Event{Kind: event.KIND_PUSH_REGISTRATION, CreatedAt: created, Tags: [][]string{
34 {"d", d}, {"relay", tenant.RelayURL()}, {"callback", "https://push.example"}, {"filter", filter},
35 }}
36 if err := event.Sign(&registration, testOwnerSecret); err != nil {
37 t.Fatal(err)
38 }
39 if _, err := tenant.store.Save(context.Background(), registration, storage.SaveOptions{Now: created}); err != nil {
40 t.Fatal(err)
41 }
42 return registration
43}
44
45func TestReplicationPushFollowupUsesInsertionCutoff(t *testing.T) {
46 _, tenant := loopbackTenant(t)
47 enablePushPlanning(t, tenant)
48 ctx := context.Background()
49 first := makeReplicationPushRegistration(t, tenant, "first", 100)
50 e := signedEvent(t, testOwnerSecret, 1, 101, nil, "push cutoff")
51 payload, err := tenant.prepareReplicationPushFollowup(ctx, e, 102)
52 if err != nil {
53 t.Fatal(err)
54 }
55 if len(payload.RegistrationIDs) != 1 || payload.RegistrationIDs[0] != first.ID {
56 t.Fatalf("captured registrations = %#v", payload.RegistrationIDs)
57 }
58 if payload.RegistrationCutoff == 0 {
59 t.Fatal("missing insertion cutoff")
60 }
61 if _, err := tenant.store.Save(ctx, e, storage.SaveOptions{Now: 102}); err != nil {
62 t.Fatal(err)
63 }
64 makeReplicationPushRegistration(t, tenant, "later", 1)
65 payload.Discover = true
66 if err := tenant.recoverReplicationPushFollowup(ctx, &payload); err != nil {
67 t.Fatal(err)
68 }
69 intents, err := tenant.prepareReplicationPushIntents(ctx, payload)
70 if err != nil {
71 t.Fatal(err)
72 }
73 if len(intents) != 1 || intents[0].Target != first.ID {
74 t.Fatalf("recovered callback intents = %#v", intents)
75 }
76}
77
78func TestReplicationPushFollowupSkipsSupersededRegistration(t *testing.T) {
79 _, tenant := loopbackTenant(t)
80 enablePushPlanning(t, tenant)
81 ctx := context.Background()
82 old := makeReplicationPushRegistration(t, tenant, "same", 100)
83 e := signedEvent(t, testOwnerSecret, 1, 101, nil, "push replacement")
84 payload, err := tenant.prepareReplicationPushFollowup(ctx, e, 102)
85 if err != nil {
86 t.Fatal(err)
87 }
88 if _, err := tenant.store.Save(ctx, e, storage.SaveOptions{Now: time.Now().Unix()}); err != nil {
89 t.Fatal(err)
90 }
91 newer := event.Event{Kind: event.KIND_PUSH_REGISTRATION, CreatedAt: 200, Tags: [][]string{
92 {"d", "same"}, {"relay", tenant.RelayURL()}, {"callback", "https://push.example"}, {"filter", `{"kinds":[1]}`},
93 }}
94 if err := event.Sign(&newer, testOwnerSecret); err != nil {
95 t.Fatal(err)
96 }
97 if _, err := tenant.store.Save(ctx, newer, storage.SaveOptions{Now: 200}); err != nil {
98 t.Fatal(err)
99 }
100 intents, err := tenant.prepareReplicationPushIntents(ctx, payload)
101 if err != nil {
102 t.Fatal(err)
103 }
104 if len(intents) != 0 {
105 t.Fatalf("superseded callback intents = %#v (old registration %s)", intents, old.ID)
106 }
107}
108
109func TestReplicationPushFollowupFailureKeepsClientEventDurable(t *testing.T) {
110 _, tenant := loopbackTenant(t)
111 enablePushPlanning(t, tenant)
112 ctx := context.Background()
113 registration := makeReplicationPushRegistration(t, tenant, "durable", 100)
114 if _, err := tenant.store.DB().ExecContext(ctx, `CREATE TRIGGER reject_replication_push_work BEFORE INSERT ON work_intents WHEN NEW.kind='callback' BEGIN SELECT RAISE(ABORT,'push work unavailable'); END`); err != nil {
115 t.Fatal(err)
116 }
117 e := signedEvent(t, testOwnerSecret, 1, 101, nil, "durable push")
118 if _, err := tenant.Publish(ctx, e, relay.Session{PubKeys: []string{e.PubKey}}); err != nil {
119 t.Fatalf("publish rejected after optional push planning failure: %v", err)
120 }
121 var stored int
122 if err := tenant.store.DB().QueryRowContext(ctx, `SELECT count(*) FROM events WHERE id=?`, e.ID).Scan(&stored); err != nil {
123 t.Fatal(err)
124 }
125 if stored != 1 {
126 t.Fatalf("stored events=%d, want 1", stored)
127 }
128 job := followupJob(t, tenant, e.ID)
129 if job.Kind != eventFollowupKind {
130 t.Fatalf("follow-up kind=%q", job.Kind)
131 }
132 if err := tenant.store.DB().QueryRowContext(ctx, `SELECT state FROM work_intents WHERE id=?`, job.ID).Scan(&job.State); err != nil {
133 t.Fatal(err)
134 }
135 if job.State != "pending" {
136 t.Fatalf("parent state=%q, want pending", job.State)
137 }
138 if _, err := tenant.store.DB().ExecContext(ctx, `DROP TRIGGER reject_replication_push_work`); err != nil {
139 t.Fatal(err)
140 }
141 parent := storage.Intent{Kind: job.Kind, EventID: job.EventID, Target: job.Target, Payload: job.Payload}
142 tenant.followups.try(ctx, &parent)
143 var parentState string
144 if err := tenant.store.DB().QueryRowContext(ctx, `SELECT state FROM work_intents WHERE id=?`, job.ID).Scan(&parentState); err != nil {
145 t.Fatal(err)
146 }
147 if parentState != "completed" {
148 t.Fatalf("recovered parent state=%q, want completed", parentState)
149 }
150 var callbackState string
151 if err := tenant.store.DB().QueryRowContext(ctx, `SELECT state FROM work_intents WHERE kind='callback' AND event_id=? AND target=?`, e.ID, registration.ID).Scan(&callbackState); err != nil {
152 t.Fatal(err)
153 }
154 if callbackState != "pending" {
155 t.Fatalf("derived callback state=%q, want pending", callbackState)
156 }
157 if _, err := tenant.store.DB().ExecContext(ctx, `UPDATE work_intents SET state='completed' WHERE kind='callback' AND event_id=? AND target=?`, e.ID, registration.ID); err != nil {
158 t.Fatal(err)
159 }
160 tenant.followups.try(ctx, &parent)
161 if err := tenant.store.DB().QueryRowContext(ctx, `SELECT state FROM work_intents WHERE kind='callback' AND event_id=? AND target=?`, e.ID, registration.ID).Scan(&callbackState); err != nil {
162 t.Fatal(err)
163 }
164 if callbackState != "completed" {
165 t.Fatalf("recovered completed callback state=%q, want completed", callbackState)
166 }
167}
168
169func TestReplicationPushFollowupSkipsRevokedPolicy(t *testing.T) {
170 _, tenant := loopbackTenant(t)
171 enablePushPlanning(t, tenant)
172 ctx := context.Background()
173 registration := makeReplicationPushRegistration(t, tenant, "revoked", 100)
174 e := signedEvent(t, testOwnerSecret, 1, 101, nil, "revoked push")
175 payload, err := tenant.prepareReplicationPushFollowup(ctx, e, 102)
176 if err != nil {
177 t.Fatal(err)
178 }
179 if _, err := tenant.store.Save(ctx, e, storage.SaveOptions{Now: 102}); err != nil {
180 t.Fatal(err)
181 }
182 payload.RegistrationIDs = []string{registration.ID}
183 p := tenant.Policy()
184 p.Features.Push = false
185 p.PushCallbacks = nil
186 if err := tenant.applyPolicy(ctx, p); err != nil {
187 t.Fatal(err)
188 }
189 intents, err := tenant.planReplicationPushFollowup(ctx, payload)
190 if err != nil {
191 t.Fatal(err)
192 }
193 if len(intents) != 0 {
194 t.Fatalf("revoked push intents=%d, want 0", len(intents))
195 }
196}
197
198func TestReplicationPushFollowupGitReleaseUsesCapturedRegistrations(t *testing.T) {
199 // A pending source retains its client snapshot in the parent follow-up;
200 // releasedPush must recover that snapshot instead of rediscovering later
201 // registrations.
202 _, tenant := loopbackTenant(t)
203 enablePushPlanning(t, tenant)
204 ctx := context.Background()
205 first := makeReplicationPushRegistration(t, tenant, "git-first", 100, event.KIND_REPO_STATE)
206 e := signedEvent(t, testOwnerSecret, event.KIND_REPO_STATE, 101, nil, "pending git")
207 payload, err := tenant.prepareReplicationPushFollowup(ctx, e, 102)
208 if err != nil {
209 t.Fatal(err)
210 }
211 followup := tenant.followups.prepareWithPush(ctx, e, 102, false, &payload)
212 if followup == nil {
213 t.Fatal("missing pending git follow-up")
214 }
215 if _, err := tenant.store.Save(ctx, e, storage.SaveOptions{Now: 102, Intents: []storage.Intent{*followup}}); err != nil {
216 t.Fatal(err)
217 }
218 makeReplicationPushRegistration(t, tenant, "git-later", 1, event.KIND_REPO_STATE)
219 released, err := tenant.followups.releasedPush(ctx, e.ID)
220 if err != nil {
221 t.Fatal(err)
222 }
223 if released == nil || len(released.RegistrationIDs) != 1 || released.RegistrationIDs[0] != first.ID {
224 t.Fatalf("released push snapshot=%#v", released)
225 }
226 intents, err := tenant.prepareReplicationPushIntents(ctx, *released)
227 if err != nil {
228 t.Fatal(err)
229 }
230 if len(intents) != 1 || intents[0].Target != first.ID {
231 t.Fatalf("released push intents=%#v", intents)
232 }
233}

About

ownerc01503e8035f
visibilitypublic
headd730679639bf

Maintainers

  • c01503e8035f owner
  • ba1b5beed5b5 agent hermes

Clone

  • https://012.run/npub1cq2s86qrtadf36eqq3epm38wmhdynrrcwxf84eldk3f94dkv7ttsrulrz5/tinyrelay.git

Refs