tinyrelayfile
tinyrelay/internal/relay/relay_test.go/
internal/relay/relay_test.go raw
1package relay
2
3import (
4 "context"
5 "encoding/json"
6 "net/http"
7 "net/http/httptest"
8 "strings"
9 "testing"
10 "time"
11
12 "github.com/FelineStateMachine/tinyrelay/internal/event"
13 "github.com/FelineStateMachine/tinyrelay/internal/syncprotocol"
14 "github.com/coder/websocket"
15)
16
17type testBackend struct{}
18
19func (testBackend) Publish(context.Context, event.Event, Session) (string, error) { return "", nil }
20func (testBackend) Query(context.Context, []event.Filter, Session) ([]event.Event, error) {
21 return nil, nil
22}
23func (testBackend) Count(context.Context, []event.Filter, Session) (any, error) { return 0, nil }
24
25type resultBackend struct{ reason string }
26
27func (b resultBackend) Publish(context.Context, event.Event, Session) (string, error) {
28 return b.reason, nil
29}
30func (resultBackend) Query(context.Context, []event.Filter, Session) ([]event.Event, error) {
31 return nil, nil
32}
33func (resultBackend) Count(context.Context, []event.Filter, Session) (any, error) { return 0, nil }
34
35type filterBackend struct{ resultBackend }
36
37func (filterBackend) CanReadFilter(event.Event, Session, *event.Filter) bool { return false }
38
39type gatedBackend struct {
40 testBackend
41 queryStarted chan struct{}
42 allowQuery chan struct{}
43 publish chan struct{}
44 syncItems []syncprotocol.Item
45}
46
47func (b *gatedBackend) Query(context.Context, []event.Filter, Session) ([]event.Event, error) {
48 close(b.queryStarted)
49 <-b.allowQuery
50 return nil, nil
51}
52func (b *gatedBackend) Publish(context.Context, event.Event, Session) (string, error) {
53 b.publish <- struct{}{}
54 return "", nil
55}
56func (b *gatedBackend) Sync(context.Context, event.Filter, Session) ([]syncprotocol.Item, error) {
57 return b.syncItems, nil
58}
59
60func TestPendingQueueRejectsSlowConsumerBeforeUnboundedGrowth(t *testing.T) {
61 r := New(testBackend{}, Config{MaxPendingBytes: 4})
62 c := newClient(r, nil, httptest.NewRequest("GET", "http://relay.example/", nil))
63 if err := c.enqueue("1234"); err != nil {
64 t.Fatalf("first message: %v", err)
65 }
66 if err := c.enqueue("5"); err == nil {
67 t.Fatal("expected slow-consumer rejection")
68 }
69 c.cancel()
70}
71
72func TestPublishDoesNotRebroadcastDuplicate(t *testing.T) {
73 r := New(resultBackend{reason: "duplicate: already have this event"}, Config{})
74 c := newClient(r, nil, httptest.NewRequest("GET", "http://relay.example/", nil))
75 c.subs["sub"] = subscription{filters: []event.Filter{{Kinds: []int{1}}}}
76 r.clientsMu.Lock()
77 r.clients[c] = struct{}{}
78 r.clientsMu.Unlock()
79 _, err := r.Publish(context.Background(), event.Event{ID: strings.Repeat("a", 64), Kind: 1}, Session{})
80 if err != nil {
81 t.Fatal(err)
82 }
83 c.mu.Lock()
84 defer c.mu.Unlock()
85 if len(c.queue) != 0 {
86 t.Fatalf("duplicate publication was rebroadcast: %d queued messages", len(c.queue))
87 }
88}
89
90func TestFanoutUsesFilterAwarePrivacyGate(t *testing.T) {
91 r := New(filterBackend{}, Config{})
92 c := newClient(r, nil, httptest.NewRequest("GET", "http://relay.example/", nil))
93 c.subs["sub"] = subscription{filters: []event.Filter{{Kinds: []int{1}}}}
94 r.clientsMu.Lock()
95 r.clients[c] = struct{}{}
96 r.clientsMu.Unlock()
97 r.BroadcastGenerated(event.Event{ID: strings.Repeat("a", 64), Kind: 1})
98 c.mu.Lock()
99 defer c.mu.Unlock()
100 if len(c.queue) != 0 {
101 t.Fatal("filter-aware reader gate was bypassed")
102 }
103}
104
105func TestPolicyChangeClosesSubscriptionsAndSyncSessions(t *testing.T) {
106 r := New(testBackend{}, Config{})
107 c := newClient(r, nil, httptest.NewRequest("GET", "http://relay.example/", nil))
108 c.subs["sub"] = subscription{filters: []event.Filter{{Kinds: []int{1}}}}
109 c.syncs["sync"] = &syncprotocol.Session{}
110 r.clientsMu.Lock()
111 r.clients[c] = struct{}{}
112 r.clientsMu.Unlock()
113 r.CloseSubscriptions("blocked: revoked")
114 c.mu.Lock()
115 defer c.mu.Unlock()
116 if len(c.subs) != 0 || len(c.syncs) != 0 {
117 t.Fatal("policy change retained a live subscription or sync session")
118 }
119 if len(c.queue) != 2 || !strings.Contains(string(c.queue[1].data), `"NEG-CLOSE","sync"`) {
120 t.Fatalf("policy change did not close both protocol sessions: %+v", c.queue)
121 }
122}
123
124func TestApplyACLChangeInvalidatesSyncUnderPublicationFence(t *testing.T) {
125 r := New(testBackend{}, Config{})
126 c := newClient(r, nil, httptest.NewRequest("GET", "http://relay.example/", nil))
127 c.syncs["sync"] = &syncprotocol.Session{}
128 r.clientsMu.Lock()
129 r.clients[c] = struct{}{}
130 r.clientsMu.Unlock()
131 called := false
132 if _, err := r.ApplyACLChange(context.Background(), func(context.Context) (any, error) {
133 called = true
134 return "ok", nil
135 }, "blocked: membership revoked"); err != nil {
136 t.Fatal(err)
137 }
138 if !called {
139 t.Fatal("ACL mutation callback did not run")
140 }
141 c.mu.Lock()
142 defer c.mu.Unlock()
143 if len(c.syncs) != 0 || len(c.queue) != 1 || !strings.Contains(string(c.queue[0].data), `"NEG-CLOSE","sync"`) {
144 t.Fatalf("sync session survived ACL mutation: %+v", c.queue)
145 }
146}
147
148func TestAcceptedMembershipEventInvalidatesSyncBeforeFanout(t *testing.T) {
149 r := New(testBackend{}, Config{})
150 c := newClient(r, nil, httptest.NewRequest("GET", "http://relay.example/", nil))
151 c.syncs["sync"] = &syncprotocol.Session{}
152 c.subs["sub"] = subscription{filters: []event.Filter{{Kinds: []int{event.KIND_LEAVE}}}}
153 r.clientsMu.Lock()
154 r.clients[c] = struct{}{}
155 r.clientsMu.Unlock()
156 _, err := r.Publish(context.Background(), event.Event{ID: strings.Repeat("a", 64), Kind: event.KIND_LEAVE}, Session{})
157 if err != nil {
158 t.Fatal(err)
159 }
160 c.mu.Lock()
161 defer c.mu.Unlock()
162 if len(c.syncs) != 0 || len(c.subs) != 0 {
163 t.Fatal("accepted membership event retained stale transport state")
164 }
165 if len(c.queue) < 2 || !queueContains(c.queue, `"NEG-CLOSE","sync"`) {
166 t.Fatalf("membership revocation did not close sync before fanout: %+v", c.queue)
167 }
168}
169
170func queueContains(queue []outbound, value string) bool {
171 for _, item := range queue {
172 if strings.Contains(string(item.data), value) {
173 return true
174 }
175 }
176 return false
177}
178
179func TestZeroMessageLimitIsUnlimited(t *testing.T) {
180 r := New(testBackend{}, Config{MaxMessageBytes: 0})
181 if r.cfg.MaxMessageBytes != 0 {
182 t.Fatalf("zero message limit became %d", r.cfg.MaxMessageBytes)
183 }
184 if got := r.cfg.withDefaults().MaxMessageBytes; got != 0 {
185 t.Fatalf("zero message limit became %d", got)
186 }
187}
188
189func TestIdlePingPongKeepsSessionAliveForLaterTraffic(t *testing.T) {
190 r := New(testBackend{}, Config{ReadLimit: 20 * time.Millisecond, PingInterval: 5 * time.Millisecond})
191 server := httptest.NewServer(http.HandlerFunc(r.HandleHTTP))
192 defer server.Close()
193 conn, _, err := websocket.Dial(context.Background(), "ws"+strings.TrimPrefix(server.URL, "http"), nil)
194 if err != nil {
195 t.Fatal(err)
196 }
197 defer conn.Close(websocket.StatusNormalClosure, "done")
198
199 readCtx, cancel := context.WithCancel(context.Background())
200 defer cancel()
201 texts := make(chan []byte, 8)
202 readErr := make(chan error, 1)
203 go func() {
204 for {
205 typ, data, err := conn.Read(readCtx)
206 if err != nil {
207 readErr <- err
208 return
209 }
210 if typ == websocket.MessageText {
211 texts <- data
212 }
213 }
214 }()
215 select {
216 case <-texts:
217 case err := <-readErr:
218 t.Fatalf("read AUTH: %v", err)
219 case <-time.After(time.Second):
220 t.Fatal("timed out waiting for AUTH")
221 }
222 time.Sleep(50 * time.Millisecond)
223 if err := conn.Write(context.Background(), websocket.MessageText, []byte(`["REQ","idle",{}]`)); err != nil {
224 t.Fatalf("write REQ after idle: %v", err)
225 }
226 select {
227 case data := <-texts:
228 if !strings.Contains(string(data), `"EOSE","idle"`) {
229 t.Fatalf("response after idle = %s", data)
230 }
231 case err := <-readErr:
232 t.Fatalf("session closed during healthy idle: %v", err)
233 case <-time.After(time.Second):
234 t.Fatal("timed out waiting for EOSE after idle")
235 }
236 secret := strings.Repeat("1", 64)
237 e := event.Event{Kind: 1, CreatedAt: time.Now().Unix(), Content: "after idle"}
238 if err := event.Sign(&e, secret); err != nil {
239 t.Fatal(err)
240 }
241 raw, _ := json.Marshal([]any{"EVENT", e})
242 if err := conn.Write(context.Background(), websocket.MessageText, raw); err != nil {
243 t.Fatalf("write EVENT after idle: %v", err)
244 }
245 select {
246 case data := <-texts:
247 if !strings.Contains(string(data), `"OK","`+e.ID+`",true`) {
248 t.Fatalf("publish response after idle = %s", data)
249 }
250 case err := <-readErr:
251 t.Fatalf("session closed before publish response: %v", err)
252 case <-time.After(time.Second):
253 t.Fatal("timed out waiting for publish response after idle")
254 }
255}
256
257func TestRemoteIPStripsPortWithoutTrustingForwardedHeaders(t *testing.T) {
258 if got := remoteIP("192.0.2.10:7447"); got != "192.0.2.10" {
259 t.Fatalf("remote IP = %q", got)
260 }
261 if got := remoteIP("[2001:db8::1]:7447"); got != "2001:db8::1" {
262 t.Fatalf("IPv6 remote IP = %q", got)
263 }
264}
265
266func TestAuthUsesConnectionRelayURL(t *testing.T) {
267 r := New(testBackend{}, Config{
268 RelayURL: "wss://canonical.example/tenant",
269 RequestRelayURL: func(*http.Request) string {
270 return "wss://custom.example/tenant"
271 },
272 })
273 c := newClient(r, nil, httptest.NewRequest("GET", "https://custom.example/tenant", nil))
274 secret := strings.Repeat("1", 64)
275 auth := event.Event{CreatedAt: time.Now().Unix(), Kind: event.KIND_AUTH, Tags: [][]string{{"relay", "wss://custom.example/tenant/"}, {"challenge", c.challenge}}, Content: ""}
276 if err := event.Sign(&auth, secret); err != nil {
277 t.Fatal(err)
278 }
279 c.handleAuth([]json.RawMessage{mustRaw(auth)})
280 c.mu.Lock()
281 if !strings.Contains(string(c.queue[len(c.queue)-1].data), `"OK","`+auth.ID+`",true`) {
282 t.Fatalf("custom relay AUTH was rejected: %s", c.queue[len(c.queue)-1].data)
283 }
284 c.mu.Unlock()
285
286 bad := auth
287 bad.Tags = [][]string{{"relay", "wss://other.example"}, {"challenge", c.challenge}}
288 if err := event.Sign(&bad, secret); err != nil {
289 t.Fatal(err)
290 }
291 c.handleAuth([]json.RawMessage{mustRaw(bad)})
292 c.mu.Lock()
293 defer c.mu.Unlock()
294 if !strings.Contains(string(c.queue[len(c.queue)-1].data), `"OK","`+bad.ID+`",false`) {
295 t.Fatalf("mismatched relay AUTH was accepted: %s", c.queue[len(c.queue)-1].data)
296 }
297}
298
299func mustRaw(v any) json.RawMessage {
300 b, _ := json.Marshal(v)
301 return b
302}
303
304func TestNormalizeRelayIgnoresTrailingSlashOnly(t *testing.T) {
305 if normalizeRelay("WSS://Relay.Example:8448/path///") != "relay.example" {
306 t.Fatal("relay URL normalization should compare the relay host")
307 }
308}
309
310func TestREQQueryAndPublishShareTheHistoricalLiveFence(t *testing.T) {
311 b := &gatedBackend{queryStarted: make(chan struct{}), allowQuery: make(chan struct{}), publish: make(chan struct{}, 1)}
312 r := New(b, Config{})
313 c := newClient(r, nil, httptest.NewRequest("GET", "http://relay.example/", nil))
314 requestDone := make(chan struct{})
315 go func() {
316 c.handleRequest("REQ", []json.RawMessage{json.RawMessage(`"sub"`), json.RawMessage(`{}`)})
317 close(requestDone)
318 }()
319 <-b.queryStarted
320 publishDone := make(chan struct{})
321 go func() {
322 r.mu.Lock()
323 _, _ = r.backend.Publish(context.Background(), event.Event{}, c.snapshot())
324 r.mu.Unlock()
325 close(publishDone)
326 }()
327 select {
328 case <-b.publish:
329 t.Fatal("publish crossed the historical query fence")
330 case <-time.After(20 * time.Millisecond):
331 }
332 close(b.allowQuery)
333 <-requestDone
334 <-publishDone
335 c.cancel()
336}
337
338func TestCommitGeneratedSharesHistoricalLiveFence(t *testing.T) {
339 b := &gatedBackend{queryStarted: make(chan struct{}), allowQuery: make(chan struct{}), publish: make(chan struct{}, 1)}
340 r := New(b, Config{})
341 c := newClient(r, nil, httptest.NewRequest("GET", "http://relay.example/", nil))
342 queryDone := make(chan struct{})
343 go func() {
344 c.handleRequest("REQ", []json.RawMessage{json.RawMessage(`"sub"`), json.RawMessage(`{}`)})
345 close(queryDone)
346 }()
347 <-b.queryStarted
348 commitDone := make(chan struct{})
349 go func() {
350 _, _ = r.CommitGenerated(context.Background(), func(context.Context) (event.Event, error) {
351 close(commitDone)
352 return event.Event{}, nil
353 })
354 }()
355 select {
356 case <-commitDone:
357 t.Fatal("generated commit crossed the historical query fence")
358 case <-time.After(20 * time.Millisecond):
359 }
360 close(b.allowQuery)
361 <-queryDone
362 select {
363 case <-commitDone:
364 case <-time.After(time.Second):
365 t.Fatal("generated commit did not proceed after query")
366 }
367 c.cancel()
368}
369
370func TestNegOpenUsesAuthorizedSnapshotAndReportsProtocolResult(t *testing.T) {
371 initiator, err := syncprotocol.NewSession([]syncprotocol.Item{{ID: strings.Repeat("a", 64), Timestamp: 1}}, true)
372 if err != nil {
373 t.Fatal(err)
374 }
375 message, err := initiator.Start()
376 if err != nil {
377 t.Fatal(err)
378 }
379 b := &gatedBackend{syncItems: []syncprotocol.Item{{ID: strings.Repeat("b", 64), Timestamp: 2}}}
380 r := New(b, Config{})
381 c := newClient(r, nil, httptest.NewRequest("GET", "http://relay.example/", nil))
382 c.handleSync("NEG-OPEN", []json.RawMessage{json.RawMessage(`"sync"`), json.RawMessage(`{}`), json.RawMessage(`"` + message + `"`)})
383 c.mu.Lock()
384 defer c.mu.Unlock()
385 if c.syncs["sync"] == nil {
386 t.Fatal("NEG-OPEN did not retain a session after a non-empty response")
387 }
388 if len(c.queue) == 0 || !strings.Contains(string(c.queue[len(c.queue)-1].data), `"NEG-MSG"`) {
389 t.Fatal("NEG-OPEN did not enqueue NEG-MSG")
390 }
391}

About

ownerc01503e8035f
visibilitypublic
headd730679639bf

Maintainers

  • c01503e8035f owner
  • ba1b5beed5b5 agent hermes

Clone

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

Refs