server_test.mx raw
1 // Package server tests exercise the relay coordinator's pure logic: the
2 // subscription registry and its caps, envelope peeking, the IP/CORS/NIP-11
3 // decisions, the worker-pool bookkeeping (driven over constructed channels,
4 // with real spawn domains only for startup/respawn), the async reaper, and the
5 // database reply dispatch. Anything that needs a live socket is out of scope.
6 package server
7
8 import (
9 "bytes"
10 "testing"
11 "time"
12
13 "git.smesh.lol/nostr/pkg/envelope"
14 "git.smesh.lol/nostr/pkg/event"
15 "git.smesh.lol/nostr/pkg/kind"
16 "git.smesh.lol/nostr/pkg/signer/p8k"
17 "git.smesh.lol/nostr/pkg/tag"
18 "git.smesh.lol/morly/pkg/pool"
19 "git.smesh.lol/morly/pkg/relay/config"
20 "git.smesh.lol/morly/pkg/relay/ratelimit"
21 "git.smesh.lol/morly/pkg/relay/tree"
22 "git.smesh.lol/morly/pkg/relay/wire"
23 "git.smesh.lol/morly/pkg/transport"
24 )
25
26 // --- helpers ---
27
28 // stCfg is a minimal config: no relay URL (ungated), no auth, no CORS origins.
29 func stCfg() (c *config.C) {
30 c = &config.C{}
31 c.AppName = "musiquay-test"
32 c.ACLMode = "none"
33 c.NIP70Enforce = false
34 c.NegentropyEnabled = false
35 c.MaxSubscriptions = 10000
36 c.QueryResultLimit = 256
37 c.MaxConnPerIP = 100
38 return
39 }
40
41 // stServer builds a Server with no spawned domains and an empty transport.
42 // transport.New stores the handler; nil never gets called.
43 func stServer(c *config.C) (s *Server) {
44 s = &Server{
45 cfg: c,
46 conns: map[int32]*cstate{},
47 dbPending: map[uint32]int32{},
48 pendingReq: map[uint32]int32{},
49 asyncPending: map[uint32]asyncHTTPEntry{},
50 t: transport.New(nil, c.MaxConnPerIP),
51 }
52 return
53 }
54
55 // stFakeDB is a DB handle whose channels are buffered and whose done channel
56 // stays open, so Valid() is true and dbSend never blocks.
57 func stFakeDB() (d *DB) {
58 d = &DB{
59 in: chan tree.Request{8},
60 out: chan tree.Response{8},
61 ready: chan struct{}{8},
62 done: chan struct{}{},
63 }
64 return
65 }
66
67 func stSignedEvent(t *testing.T, k uint16, tags *tag.S, content string) (ev *event.E) {
68 t.Helper()
69 sg := p8k.MustNew()
70 if gerr := sg.Generate(); gerr != nil {
71 t.Fatal(gerr)
72 return
73 }
74 ev = event.New()
75 ev.CreatedAt = time.Now().Unix()
76 ev.Kind = k
77 ev.Tags = tags
78 ev.Content = []byte(content)
79 if serr := ev.Sign(sg); serr != nil {
80 t.Fatal(serr)
81 return
82 }
83 return
84 }
85
86 func stEventMsg(ev *event.E) (msg []byte) {
87 es := &envelope.EventSubmission{E: ev}
88 return es.Marshal(nil)
89 }
90
91 // stHas is a local substring scan. bytes.Contains/Index can miss a needle in a
92 // long haystack in this compiler, so the coverage tests do not trust them.
93 func stHas(hay, needle []byte) (found bool) {
94 if len(needle) == 0 {
95 return true
96 }
97 if len(needle) > len(hay) {
98 return false
99 }
100 var i int32
101 var j int32
102 var match bool
103 for i = 0; i+len(needle) <= len(hay); i++ {
104 match = true
105 for j = 0; j < len(needle); j++ {
106 if hay[i+j] != needle[j] {
107 match = false
108 break
109 }
110 }
111 if match {
112 return true
113 }
114 }
115 return false
116 }
117
118 func stTryProxyReq(ch chan wire.ProxyRequest) (r wire.ProxyRequest, ok bool) {
119 select {
120 case v := <-ch:
121 r = v
122 ok = true
123 default:
124 }
125 return
126 }
127
128 func stTryIngestReq(ch chan wire.IngestRequest) (r wire.IngestRequest, ok bool) {
129 select {
130 case v := <-ch:
131 r = v
132 ok = true
133 default:
134 }
135 return
136 }
137
138 func stTryDBReq(ch chan tree.Request) (r tree.Request, ok bool) {
139 select {
140 case v := <-ch:
141 r = v
142 ok = true
143 default:
144 }
145 return
146 }
147
148 // stIngestSrv builds an ingest pool with constructed channels. The done
149 // channels stay open so workerAlive reports every worker alive.
150 func stIngestSrv(n int32) (s *Server) {
151 s = stServer(stCfg())
152 // push, not []chan T{:n}: the inline slice-of-channel literal is
153 // miscompiled to an empty slice, so the pool is grown element by element.
154 s.workers = pool.NewPool(n)
155 for i := int32(0); i < n; i++ {
156 s.workerIn = push(s.workerIn, chan wire.IngestRequest{4})
157 s.workerOut = push(s.workerOut, chan wire.IngestResponse{4})
158 s.workerDone = push(s.workerDone, chan struct{}{})
159 s.workerReady = push(s.workerReady, chan struct{}{4})
160 }
161 s.db = stFakeDB()
162 return
163 }
164
165 // stProxySrv builds a proxy pool with constructed channels.
166 func stProxySrv(n int32) (s *Server) {
167 s = stServer(stCfg())
168 s.proxyPool = pool.NewPool(n)
169 s.proxyBusyTime = []int64{:n}
170 for i := int32(0); i < n; i++ {
171 s.proxyIn = push(s.proxyIn, chan wire.ProxyRequest{4})
172 s.proxyOut = push(s.proxyOut, chan wire.ProxyResponse{4})
173 s.proxyDone = push(s.proxyDone, chan struct{}{})
174 }
175 return
176 }
177
178 // stBlossomSrv builds a blossom pool with constructed channels.
179 func stBlossomSrv(n int32) (s *Server) {
180 s = stServer(stCfg())
181 s.blossomPool = pool.NewPool(n)
182 for i := int32(0); i < n; i++ {
183 s.blossomIn = push(s.blossomIn, chan wire.BlossomRequest{4})
184 s.blossomOut = push(s.blossomOut, chan wire.BlossomResponse{4})
185 s.blossomDone = push(s.blossomDone, chan struct{}{})
186 }
187 return
188 }
189
190 // --- envelope peeking ---
191
192 func TestPeekEventKind(t *testing.T) {
193 ev := stSignedEvent(t, 1, nil, "hello")
194 if k := peekEventKind(stEventMsg(ev)); k != 1 {
195 t.Fatalf("peekEventKind = %d, want 1", k)
196 return
197 }
198 eph := stSignedEvent(t, 20000, nil, "eph")
199 if k := peekEventKind(stEventMsg(eph)); k != 20000 {
200 t.Fatalf("peekEventKind ephemeral = %d, want 20000", k)
201 return
202 }
203 replace := stSignedEvent(t, 30023, nil, "article")
204 if k := peekEventKind(stEventMsg(replace)); k != 30023 {
205 t.Fatalf("peekEventKind param replaceable = %d", k)
206 return
207 }
208 }
209
210 // --- worker lifecycle ---
211
212 func TestWorkerAlive(t *testing.T) {
213 d := chan struct{}{}
214 if !workerAlive(d) {
215 t.Fatal("open channel must be alive")
216 return
217 }
218 close(d)
219 if workerAlive(d) {
220 t.Fatal("closed channel must be dead")
221 return
222 }
223 }
224
225 // --- URL matching ---
226
227 func TestStripScheme(t *testing.T) {
228 if string(stripScheme([]byte("wss://relay.example.com"))) != "relay.example.com" {
229 t.Fatalf("wss strip = %s", stripScheme([]byte("wss://relay.example.com")))
230 return
231 }
232 if string(stripScheme([]byte("ws://relay.example.com"))) != "relay.example.com" {
233 t.Fatalf("ws strip = %s", stripScheme([]byte("ws://relay.example.com")))
234 return
235 }
236 if string(stripScheme([]byte("https://relay.example.com"))) != "https://relay.example.com" {
237 t.Fatalf("https must be untouched, got %s", stripScheme([]byte("https://relay.example.com")))
238 return
239 }
240 if string(stripScheme([]byte("relay.example.com"))) != "relay.example.com" {
241 t.Fatal("bare host must be untouched")
242 return
243 }
244 }
245
246 func TestRelayURLMatch(t *testing.T) {
247 if !relayURLMatch([]byte("wss://relay.example.com"), []byte("wss://relay.example.com")) {
248 t.Fatal("identical URLs must match")
249 return
250 }
251 if !relayURLMatch([]byte("wss://relay.example.com"), []byte("relay.example.com")) {
252 t.Fatal("scheme difference must match")
253 return
254 }
255 if !relayURLMatch([]byte("wss://relay.example.com/"), []byte("relay.example.com")) {
256 t.Fatal("trailing slash must match")
257 return
258 }
259 if !relayURLMatch([]byte("WSS://RELAY.EXAMPLE.COM"), []byte("relay.example.com")) {
260 t.Fatal("case difference must match")
261 return
262 }
263 if !relayURLMatch([]byte("ws://relay.example.com"), []byte("wss://relay.example.com")) {
264 t.Fatal("ws vs wss must match after scheme strip")
265 return
266 }
267 if relayURLMatch([]byte("wss://a.example.com"), []byte("b.example.com")) {
268 t.Fatal("different hosts must not match")
269 return
270 }
271 if relayURLMatch([]byte("https://relay.example.com"), []byte("relay.example.com")) {
272 t.Fatal("unrecognised scheme must not be stripped")
273 return
274 }
275 }
276
277 func TestMakeCopyAndToLower(t *testing.T) {
278 orig := []byte("AbC")
279 cp := makeCopy(orig)
280 if !bytes.Equal(cp, []byte("AbC")) {
281 t.Fatalf("makeCopy = %s", cp)
282 return
283 }
284 cp[0] = 'x'
285 if orig[0] != 'A' {
286 t.Fatal("makeCopy must not alias the source")
287 return
288 }
289 up := toLower([]byte("AbC-9"))
290 if string(up) != "abc-9" {
291 t.Fatalf("toLower = %s", up)
292 return
293 }
294 }
295
296 // --- IP access control ---
297
298 func TestIPBlacklistAndWhitelist(t *testing.T) {
299 c := stCfg()
300 c.IPBlacklist = []string{"10.0.0.", "1.2.3.4"}
301 c.IPWhitelist = []string{"192.168."}
302 s := stServer(c)
303 if !s.ipBlacklisted("10.0.0.7") {
304 t.Fatal("10.0.0.7 must be blacklisted")
305 return
306 }
307 if !s.ipBlacklisted("1.2.3.4") {
308 t.Fatal("exact blacklist entry must match")
309 return
310 }
311 if s.ipBlacklisted("11.0.0.7") {
312 t.Fatal("11.0.0.7 must not be blacklisted")
313 return
314 }
315 if !s.ipWhitelisted("192.168.1.1") {
316 t.Fatal("192.168.1.1 must be whitelisted")
317 return
318 }
319 if s.ipWhitelisted("10.0.0.1") {
320 t.Fatal("10.0.0.1 must not be whitelisted")
321 return
322 }
323
324 empty := stServer(stCfg())
325 if empty.ipWhitelisted("192.168.1.1") {
326 t.Fatal("empty whitelist must never match")
327 return
328 }
329 if empty.ipBlacklisted("10.0.0.1") {
330 t.Fatal("empty blacklist must never match")
331 return
332 }
333 }
334
335 func TestOnAccept(t *testing.T) {
336 c := stCfg()
337 c.MaxGlobalConns = 5
338 c.IPBlacklist = []string{"9.9."}
339 s := stServer(c)
340 if !s.OnAccept(1, "1.1.1.1") {
341 t.Fatal("a normal IP under the global cap must be accepted")
342 return
343 }
344 if s.OnAccept(2, "9.9.9.9") {
345 t.Fatal("a blacklisted IP must be refused")
346 return
347 }
348 unlimited := stCfg()
349 unlimited.MaxGlobalConns = 0
350 s2 := stServer(unlimited)
351 if !s2.OnAccept(1, "1.1.1.1") {
352 t.Fatal("MaxGlobalConns=0 must mean unlimited")
353 return
354 }
355 }
356
357 func TestOnWSUpgrade(t *testing.T) {
358 c := stCfg()
359 c.MaxConnPerIP = 3
360 c.IPWhitelist = []string{"10."}
361 s := stServer(c)
362
363 wl, allow := s.OnWSUpgrade(1, "1.1.1.1", 1)
364 if wl || !allow {
365 t.Fatalf("under limit: whitelisted=%v allow=%v", wl, allow)
366 return
367 }
368 wl2, allow2 := s.OnWSUpgrade(1, "1.1.1.1", 3)
369 if wl2 || allow2 {
370 t.Fatalf("at limit: whitelisted=%v allow=%v", wl2, allow2)
371 return
372 }
373 wl3, allow3 := s.OnWSUpgrade(1, "10.1.1.1", 99)
374 if !wl3 || !allow3 {
375 t.Fatalf("whitelisted over limit must pass: whitelisted=%v allow=%v", wl3, allow3)
376 return
377 }
378
379 unlimited := stCfg()
380 unlimited.MaxConnPerIP = 0
381 s2 := stServer(unlimited)
382 wl4, allow4 := s2.OnWSUpgrade(1, "1.1.1.1", 1000)
383 if wl4 || !allow4 {
384 t.Fatal("MaxConnPerIP=0 must mean unlimited")
385 return
386 }
387 }
388
389 func TestConnHelpers(t *testing.T) {
390 s := stServer(stCfg())
391 if s.t.ConnCount() != 0 {
392 t.Fatalf("ConnCount = %d, want 0", s.t.ConnCount())
393 return
394 }
395 if s.t.ConnIP(7) != "" {
396 t.Fatalf("unknown fd ConnIP = %q, want empty", s.t.ConnIP(7))
397 return
398 }
399 if s.t.ConnIsWhitelisted(7) {
400 t.Fatal("unknown fd must not be whitelisted")
401 return
402 }
403 }
404
405 // --- CORS ---
406
407 func TestCorsHeaders(t *testing.T) {
408 c := stCfg()
409 s := stServer(c)
410
411 open := s.corsHeaders(map[string]string{})
412 if open["Access-Control-Allow-Origin"] != "*" {
413 t.Fatalf("default origin = %q", open["Access-Control-Allow-Origin"])
414 return
415 }
416 echo := s.corsHeaders(map[string]string{"origin": "https://a.example"})
417 if echo["Access-Control-Allow-Origin"] != "https://a.example" {
418 t.Fatalf("echoed origin = %q", echo["Access-Control-Allow-Origin"])
419 return
420 }
421
422 c.CORSOrigins = []string{"https://a.example"}
423 allowed := s.corsHeaders(map[string]string{"origin": "https://a.example"})
424 if allowed["Access-Control-Allow-Origin"] != "https://a.example" {
425 t.Fatal("listed origin must be allowed")
426 return
427 }
428 denied := s.corsHeaders(map[string]string{"origin": "https://evil.example"})
429 if len(denied) != 0 {
430 t.Fatalf("unlisted origin must produce no headers, got %d", len(denied))
431 return
432 }
433
434 c.CORSOrigins = []string{"*"}
435 wild := s.corsHeaders(map[string]string{"origin": "https://evil.example"})
436 if wild["Access-Control-Allow-Origin"] != "https://evil.example" {
437 t.Fatal("wildcard entry must allow any origin")
438 return
439 }
440 }
441
442 // --- NIP-11 ---
443
444 func TestNip11JSON(t *testing.T) {
445 c := stCfg()
446 c.MaxSubscriptions = 0 // exercises the 10000 default
447 s := stServer(c)
448 s.Version = "1.2.3"
449
450 buf := s.nip11JSON()
451 if !stHas(buf, []byte("\"name\":\"musiquay-test\"")) {
452 t.Fatalf("missing name in %s", buf)
453 return
454 }
455 if !stHas(buf, []byte("\"version\":\"1.2.3\"")) {
456 t.Fatal("missing version")
457 return
458 }
459 if !stHas(buf, []byte("\"supported_nips\":[1,9,11,40,42,45,50]")) {
460 t.Fatalf("unexpected nips list in %s", buf)
461 return
462 }
463 if !stHas(buf, []byte("\"max_subscriptions\":10000")) {
464 t.Fatal("max_subscriptions default must be 10000")
465 return
466 }
467 if !stHas(buf, []byte("\"auth_required\":false")) {
468 t.Fatal("auth_required must be false")
469 return
470 }
471 if !stHas(buf, []byte("\"restricted_writes\":false")) {
472 t.Fatal("restricted_writes must be false")
473 return
474 }
475 if stHas(buf, []byte("\"pubkey\"")) {
476 t.Fatal("pubkey must be omitted with no admins")
477 return
478 }
479
480 c2 := stCfg()
481 c2.NIP70Enforce = true
482 c2.NegentropyEnabled = true
483 c2.Admins = []string{"deadbeef"}
484 c2.RelayURL = "wss://relay.example.com"
485 c2.AuthRequired = true
486 c2.ACLMode = "managed"
487 c2.MaxSubscriptions = 5
488 c2.QueryResultLimit = 7
489 s2 := stServer(c2)
490 s2.Version = "9"
491 buf2 := s2.nip11JSON()
492 if !stHas(buf2, []byte(",70")) {
493 t.Fatal("NIP-70 must be advertised")
494 return
495 }
496 if !stHas(buf2, []byte(",77")) {
497 t.Fatal("NIP-77 must be advertised")
498 return
499 }
500 if !stHas(buf2, []byte("\"pubkey\":\"deadbeef\"")) {
501 t.Fatal("admin pubkey must be advertised")
502 return
503 }
504 if !stHas(buf2, []byte("\"contact\":\"wss://relay.example.com\"")) {
505 t.Fatal("relay URL must be the contact")
506 return
507 }
508 if !stHas(buf2, []byte("\"max_subscriptions\":5")) {
509 t.Fatal("max_subscriptions must be the configured value")
510 return
511 }
512 if !stHas(buf2, []byte("\"max_limit\":7")) {
513 t.Fatal("max_limit must be the query result limit")
514 return
515 }
516 if !stHas(buf2, []byte("\"auth_required\":true")) {
517 t.Fatal("auth_required must be true")
518 return
519 }
520 if !stHas(buf2, []byte("\"restricted_writes\":true")) {
521 t.Fatal("restricted_writes must be true")
522 return
523 }
524 }
525
526 // --- async reaper ---
527
528 func TestAsyncReapStuck(t *testing.T) {
529 s := stServer(stCfg())
530 now := time.Now().UnixNano()
531 s.asyncPending[1] = asyncHTTPEntry{connFD: 10, createdAt: now - 120_000_000_000}
532 s.asyncPending[2] = asyncHTTPEntry{connFD: 11, createdAt: now}
533 s.asyncPending[3] = asyncHTTPEntry{connFD: 12, createdAt: 0}
534 s.asyncReapStuck()
535
536 if _, ok := s.asyncPending[1]; ok {
537 t.Fatal("an entry older than 60s must be reaped")
538 return
539 }
540 if _, ok := s.asyncPending[2]; !ok {
541 t.Fatal("a fresh entry must survive")
542 return
543 }
544 if _, ok := s.asyncPending[3]; !ok {
545 t.Fatal("createdAt=0 must not be treated as stale")
546 return
547 }
548 if len(s.asyncPending) != 2 {
549 t.Fatalf("asyncPending size = %d, want 2", len(s.asyncPending))
550 return
551 }
552 }
553
554 // --- proxy queue bookkeeping ---
555
556 func TestDrainProxyQueue(t *testing.T) {
557 s := stProxySrv(2)
558 s.proxyQueue = push(s.proxyQueue, pendingProxy{
559 connFD: 7, connClose: true, req: wire.ProxyRequest{ReqID: 1, URL: []byte("https://a")},
560 })
561 s.proxyQueue = push(s.proxyQueue, pendingProxy{
562 connFD: 8, connClose: false, req: wire.ProxyRequest{ReqID: 2, URL: []byte("https://b")},
563 })
564 s.drainProxyQueue()
565
566 if len(s.proxyQueue) != 0 {
567 t.Fatalf("queue must drain, %d left", len(s.proxyQueue))
568 return
569 }
570 if len(s.asyncPending) != 2 {
571 t.Fatalf("asyncPending = %d, want 2", len(s.asyncPending))
572 return
573 }
574 if !s.proxyPool.Busy[0] || !s.proxyPool.Busy[1] {
575 t.Fatal("both workers must be marked busy")
576 return
577 }
578 if s.proxyBusyTime[0] == 0 || s.proxyBusyTime[1] == 0 {
579 t.Fatal("busy time must be stamped")
580 return
581 }
582 r0, ok0 := stTryProxyReq(s.proxyIn[0])
583 if !ok0 || r0.ReqID != 1 {
584 t.Fatalf("worker 0 request = %v ok=%v", r0.ReqID, ok0)
585 return
586 }
587 r1, ok1 := stTryProxyReq(s.proxyIn[1])
588 if !ok1 || r1.ReqID != 2 {
589 t.Fatalf("worker 1 request = %v ok=%v", r1.ReqID, ok1)
590 return
591 }
592 e, eok := s.asyncPending[1]
593 if !eok || e.connFD != 7 || !e.connClose {
594 t.Fatal("asyncPending entry 1 must carry the pending connection")
595 return
596 }
597 }
598
599 func TestDrainProxyQueueStopsWhenBusy(t *testing.T) {
600 s := stProxySrv(1)
601 s.proxyQueue = push(s.proxyQueue, pendingProxy{connFD: 1, req: wire.ProxyRequest{ReqID: 1}})
602 s.proxyQueue = push(s.proxyQueue, pendingProxy{connFD: 2, req: wire.ProxyRequest{ReqID: 2}})
603 s.drainProxyQueue()
604
605 if len(s.proxyQueue) != 1 {
606 t.Fatalf("one entry must stay queued, %d left", len(s.proxyQueue))
607 return
608 }
609 if !s.proxyPool.Busy[0] {
610 t.Fatal("the single worker must be busy")
611 return
612 }
613 if s.proxyQueue[0].req.ReqID != 2 {
614 t.Fatalf("the queued entry must be req 2, got %d", s.proxyQueue[0].req.ReqID)
615 return
616 }
617 }
618
619 func TestDoDispatchProxyBranches(t *testing.T) {
620 s := stProxySrv(1)
621 s.selfHost = "self.example/"
622
623 if !s.doDispatchProxy(1, "/proxy/", map[string]string{}) {
624 t.Fatal("empty target must be handled synchronously")
625 return
626 }
627 if len(s.asyncPending) != 0 {
628 t.Fatal("empty target must not register async state")
629 return
630 }
631
632 if !s.doDispatchProxy(2, "/proxy/self.example/foo", map[string]string{}) {
633 t.Fatal("self-host target must be handled synchronously")
634 return
635 }
636 if len(s.asyncPending) != 0 {
637 t.Fatal("self-host redirect must not register async state")
638 return
639 }
640
641 // Queue-full path: the only worker is busy and 64 entries are already
642 // waiting, so the request must be refused with 503.
643 s.proxyPool.Busy[0] = true
644 for i := int32(0); i < 64; i++ {
645 s.proxyQueue = push(s.proxyQueue, pendingProxy{req: wire.ProxyRequest{ReqID: uint32(i)}})
646 }
647 if !s.doDispatchProxy(3, "/proxy/https://other.example/x", map[string]string{}) {
648 t.Fatal("a refused proxy request still reports handled")
649 return
650 }
651 if len(s.proxyQueue) != 64 {
652 t.Fatalf("a full queue must not grow, size = %d", len(s.proxyQueue))
653 return
654 }
655 }
656
657 // --- ingest pool ---
658
659 func TestDispatchToWorker(t *testing.T) {
660 s := stIngestSrv(1)
661 if !s.dispatchToWorker(5, []byte("[\"EVENT\",{}]")) {
662 t.Fatal("an idle worker must accept the request")
663 return
664 }
665 if !s.workers.Busy[0] {
666 t.Fatal("the worker must be marked busy")
667 return
668 }
669 if len(s.pendingReq) != 1 {
670 t.Fatalf("pendingReq = %d, want 1", len(s.pendingReq))
671 return
672 }
673 req, ok := stTryIngestReq(s.workerIn[0])
674 if !ok || req.ReqID != 1 {
675 t.Fatalf("queued ingest request = %v ok=%v", req.ReqID, ok)
676 return
677 }
678 if s.dispatchToWorker(5, []byte("x")) {
679 t.Fatal("a busy pool must refuse dispatch")
680 return
681 }
682 }
683
684 func TestCompleteIngestResponseReject(t *testing.T) {
685 s := stIngestSrv(1)
686 fd := int32(7)
687 s.conns[fd] = &cstate{subs: map[string]*sub{}}
688 s.pendingReq[3] = fd
689
690 var r wire.IngestResponse
691 r.ReqID = 3
692 r.Verdict = wire.VerdictReject
693 r.Reason = []byte("invalid: nope")
694 s.completeIngestResponse(r)
695
696 if _, ok := s.pendingReq[3]; ok {
697 t.Fatal("a completed request must be dropped from pendingReq")
698 return
699 }
700 if len(s.dbPending) != 0 {
701 t.Fatal("a rejected event must not reach the database")
702 return
703 }
704 }
705
706 func TestCompleteIngestResponseAccept(t *testing.T) {
707 s := stIngestSrv(1)
708 fd := int32(7)
709 s.conns[fd] = &cstate{subs: map[string]*sub{}}
710 s.pendingReq[3] = fd
711
712 var r wire.IngestResponse
713 r.ReqID = 3
714 r.Verdict = wire.VerdictAccept
715 r.Bytes = []byte("[\"EVENT\",{}]")
716 s.completeIngestResponse(r)
717
718 if _, ok := s.pendingReq[3]; ok {
719 t.Fatal("an accepted request must be dropped from pendingReq")
720 return
721 }
722 if len(s.dbPending) != 1 {
723 t.Fatalf("dbPending = %d, want 1", len(s.dbPending))
724 return
725 }
726 req, got := stTryDBReq(s.db.in)
727 if !got {
728 t.Fatal("the accepted event must be forwarded to the database")
729 return
730 }
731 if req.Op != tree.OpPersist || req.ConnID != fd || !req.Verified {
732 t.Fatalf("db request op=%d conn=%d verified=%v", req.Op, req.ConnID, req.Verified)
733 return
734 }
735 }
736
737 func TestCompleteIngestResponseUnknownAndGone(t *testing.T) {
738 s := stIngestSrv(1)
739 s.conns[7] = &cstate{subs: map[string]*sub{}}
740
741 var unknown wire.IngestResponse
742 unknown.ReqID = 99
743 unknown.Verdict = wire.VerdictAccept
744 s.completeIngestResponse(unknown)
745 if len(s.dbPending) != 0 {
746 t.Fatal("an unknown ReqID must be ignored")
747 return
748 }
749
750 // Known request but the connection is gone: drop it without forwarding.
751 s.pendingReq[4] = 42
752 var gone wire.IngestResponse
753 gone.ReqID = 4
754 gone.Verdict = wire.VerdictAccept
755 s.completeIngestResponse(gone)
756 if _, ok := s.pendingReq[4]; ok {
757 t.Fatal("a request for a closed connection must be dropped")
758 return
759 }
760 if len(s.dbPending) != 0 {
761 t.Fatal("a closed connection must not reach the database")
762 return
763 }
764 }
765
766 func TestPollIngestWorkersDrainsReady(t *testing.T) {
767 s := stIngestSrv(2)
768 fd := int32(5)
769 s.conns[fd] = &cstate{subs: map[string]*sub{}}
770 s.pendingReq[9] = fd
771 s.workers.Busy[0] = true
772
773 s.workerOut[0] <- wire.IngestResponse{ReqID: 9, Verdict: wire.VerdictReject, Reason: []byte("invalid: nope")}
774 s.workerReady[0] <- struct{}{}
775 s.pollIngestWorkers()
776
777 if s.workers.Busy[0] {
778 t.Fatal("a drained worker must be idle again")
779 return
780 }
781 if _, ok := s.pendingReq[9]; ok {
782 t.Fatal("the drained response must clear pendingReq")
783 return
784 }
785 if s.workers.Busy[1] {
786 t.Fatal("the untouched worker must stay idle")
787 return
788 }
789 }
790
791 // --- proxy responses ---
792
793 func TestCompleteProxyResponseStatusBranches(t *testing.T) {
794 s := stProxySrv(1)
795
796 s.asyncPending[1] = asyncHTTPEntry{connFD: 9}
797 var errResp wire.ProxyResponse
798 errResp.ReqID = 1
799 errResp.Status = -1
800 errResp.Err = []byte("boom")
801 s.completeProxyResponse(errResp)
802 if len(s.asyncPending) != 0 {
803 t.Fatal("502 branch must clear asyncPending")
804 return
805 }
806
807 s.asyncPending[2] = asyncHTTPEntry{connFD: 9}
808 var ctResp wire.ProxyResponse
809 ctResp.ReqID = 2
810 ctResp.Status = 415
811 s.completeProxyResponse(ctResp)
812 if len(s.asyncPending) != 0 {
813 t.Fatal("415 branch must clear asyncPending")
814 return
815 }
816
817 s.asyncPending[3] = asyncHTTPEntry{connFD: 9}
818 var okResp wire.ProxyResponse
819 okResp.ReqID = 3
820 okResp.Status = 200
821 okResp.ContentType = []byte("image/png")
822 okResp.Body = []byte("BODY")
823 s.completeProxyResponse(okResp)
824 if len(s.asyncPending) != 0 {
825 t.Fatal("2xx branch must clear asyncPending")
826 return
827 }
828
829 s.asyncPending[4] = asyncHTTPEntry{connFD: 9}
830 var upResp wire.ProxyResponse
831 upResp.ReqID = 4
832 upResp.Status = 404
833 s.completeProxyResponse(upResp)
834 if len(s.asyncPending) != 0 {
835 t.Fatal("default branch must clear asyncPending")
836 return
837 }
838
839 // An unknown ReqID is a no-op.
840 var stale wire.ProxyResponse
841 stale.ReqID = 77
842 stale.Status = 200
843 s.completeProxyResponse(stale)
844 if len(s.asyncPending) != 0 {
845 t.Fatal("an unknown proxy response must not create state")
846 return
847 }
848 }
849
850 func TestPollProxyWorkersDrains(t *testing.T) {
851 s := stProxySrv(1)
852 s.asyncPending[4] = asyncHTTPEntry{connFD: 9}
853 s.proxyPool.Busy[0] = true
854 s.proxyBusyTime[0] = time.Now().UnixNano()
855
856 s.proxyOut[0] <- wire.ProxyResponse{ReqID: 4, Status: 200, Body: []byte("b")}
857 s.pollProxyWorkers()
858
859 if s.proxyPool.Busy[0] {
860 t.Fatal("a drained proxy worker must be idle")
861 return
862 }
863 if s.proxyBusyTime[0] != 0 {
864 t.Fatal("a drained proxy worker must clear its busy timestamp")
865 return
866 }
867 if len(s.asyncPending) != 0 {
868 t.Fatal("the response must clear asyncPending")
869 return
870 }
871 }
872
873 // --- blossom responses ---
874
875 func TestCompleteBlossomResponseBranches(t *testing.T) {
876 s := stBlossomSrv(1)
877
878 s.asyncPending[1] = asyncHTTPEntry{connFD: 9}
879 var ctResp wire.BlossomResponse
880 ctResp.ReqID = 1
881 ctResp.Status = 200
882 ctResp.CT = []byte("image/png")
883 ctResp.Body = []byte("blob")
884 s.completeBlossomResponse(ctResp)
885 if len(s.asyncPending) != 0 {
886 t.Fatal("content-type branch must clear asyncPending")
887 return
888 }
889
890 s.asyncPending[2] = asyncHTTPEntry{connFD: 9}
891 var sizeResp wire.BlossomResponse
892 sizeResp.ReqID = 2
893 sizeResp.Status = 200
894 sizeResp.Size = 12345
895 sizeResp.Body = []byte("blob")
896 s.completeBlossomResponse(sizeResp)
897 if len(s.asyncPending) != 0 {
898 t.Fatal("size branch must clear asyncPending")
899 return
900 }
901
902 var stale wire.BlossomResponse
903 stale.ReqID = 77
904 stale.Status = 200
905 s.completeBlossomResponse(stale)
906 if len(s.asyncPending) != 0 {
907 t.Fatal("an unknown blossom response must be a no-op")
908 return
909 }
910 }
911
912 func TestPollBlossomWorkersDrains(t *testing.T) {
913 s := stBlossomSrv(1)
914 s.asyncPending[4] = asyncHTTPEntry{connFD: 9}
915 s.blossomPool.Busy[0] = true
916
917 s.blossomOut[0] <- wire.BlossomResponse{ReqID: 4, Status: 200, Body: []byte("b")}
918 s.pollBlossomWorkers()
919
920 if s.blossomPool.Busy[0] {
921 t.Fatal("a drained blossom worker must be idle")
922 return
923 }
924 if len(s.asyncPending) != 0 {
925 t.Fatal("the response must clear asyncPending")
926 return
927 }
928 }
929
930 // --- database dispatch ---
931
932 func TestDbSend(t *testing.T) {
933 s := stServer(stCfg())
934 if s.dbSend(tree.Request{Op: tree.OpPersist, ConnID: 1}) {
935 t.Fatal("dbSend without a database must fail")
936 return
937 }
938
939 s.db = stFakeDB()
940 if !s.dbSend(tree.Request{Op: tree.OpPersist, ConnID: 5}) {
941 t.Fatal("dbSend with a live database must succeed")
942 return
943 }
944 if len(s.dbPending) != 1 || s.nextDBReq != 1 {
945 t.Fatalf("dbPending=%d nextDBReq=%d", len(s.dbPending), s.nextDBReq)
946 return
947 }
948 req, ok := stTryDBReq(s.db.in)
949 if !ok || req.ReqID != 1 || req.ConnID != 5 {
950 t.Fatalf("queued db request id=%d conn=%d ok=%v", req.ReqID, req.ConnID, ok)
951 return
952 }
953
954 closed := stFakeDB()
955 close(closed.done)
956 s.db = closed
957 if s.dbSend(tree.Request{}) {
958 t.Fatal("dbSend to a stopped database must fail")
959 return
960 }
961 }
962
963 func TestPollDBDrains(t *testing.T) {
964 s := stServer(stCfg())
965 s.db = stFakeDB()
966 fd := int32(4)
967 s.conns[fd] = &cstate{subs: map[string]*sub{}}
968 s.dbPending[1] = fd
969
970 var resp tree.Response
971 resp.Op = tree.OpPersist
972 resp.ReqID = 1
973 resp.OK = true
974 resp.EventID = []byte{:32}
975 resp.Reason = []byte("saved")
976 s.db.out <- resp
977 s.db.ready <- struct{}{}
978 s.pollDB()
979
980 if _, ok := s.dbPending[1]; ok {
981 t.Fatal("the drained reply must clear dbPending")
982 return
983 }
984 }
985
986 func TestCompleteDBHistoryAndCount(t *testing.T) {
987 s := stServer(stCfg())
988 fd := int32(4)
989 cs := &cstate{subs: map[string]*sub{}}
990 cs.subs["s1"] = &sub{id: "s1"}
991 s.conns[fd] = cs
992
993 var frames [][]byte
994 frames = push(frames, []byte("[\"EVENT\",{}]"))
995 frames = push(frames, []byte("[\"EVENT\",{}]"))
996
997 var h tree.Response
998 h.Op = tree.OpHistory
999 h.ConnID = fd
1000 h.SubID = []byte("s1")
1001 h.Events = frames
1002 s.completeDB(h)
1003
1004 var unknownSub tree.Response
1005 unknownSub.Op = tree.OpHistory
1006 unknownSub.ConnID = fd
1007 unknownSub.SubID = []byte("gone")
1008 unknownSub.Events = frames
1009 s.completeDB(unknownSub)
1010
1011 var goneConn tree.Response
1012 goneConn.Op = tree.OpHistory
1013 goneConn.ConnID = 99
1014 goneConn.SubID = []byte("s1")
1015 s.completeDB(goneConn)
1016
1017 var count tree.Response
1018 count.Op = tree.OpCount
1019 count.ConnID = fd
1020 count.SubID = []byte("s1")
1021 count.Count = 5
1022 s.completeDB(count)
1023
1024 var countGone tree.Response
1025 countGone.Op = tree.OpCount
1026 countGone.ConnID = 99
1027 countGone.SubID = []byte("s1")
1028 s.completeDB(countGone)
1029
1030 var noop tree.Response
1031 noop.Op = 200
1032 s.completeDB(noop)
1033
1034 if _, still := s.conns[fd].subs["s1"]; !still {
1035 t.Fatal("a history reply must not remove a live subscription")
1036 return
1037 }
1038 }
1039
1040 func TestCompletePersistMissingAndGone(t *testing.T) {
1041 s := stServer(stCfg())
1042 s.conns[4] = &cstate{subs: map[string]*sub{}}
1043
1044 var missing tree.Response
1045 missing.Op = tree.OpPersist
1046 missing.ReqID = 88
1047 s.completeDB(missing)
1048
1049 s.dbPending[5] = 99
1050 var gone tree.Response
1051 gone.Op = tree.OpPersist
1052 gone.ReqID = 5
1053 gone.OK = true
1054 s.completeDB(gone)
1055 if _, ok := s.dbPending[5]; ok {
1056 t.Fatal("a persist reply for a closed connection must be dropped")
1057 return
1058 }
1059 }
1060
1061 // --- HTTP routing ---
1062
1063 func stFallback(method, path string, headers map[string]string, body []byte) (int32, map[string]string, []byte) {
1064 return 201, map[string]string{"X-Fallback": "1"}, []byte("fb")
1065 }
1066
1067 func TestRouteHTTP(t *testing.T) {
1068 s := stServer(stCfg())
1069 s.Version = "9"
1070
1071 status, h, body := s.routeHTTP(1, "GET", "/health", map[string]string{}, nil)
1072 if status != 200 || string(body) != "ok" {
1073 t.Fatalf("health = %d %s", status, body)
1074 return
1075 }
1076 if h["Content-Type"] != "text/plain" {
1077 t.Fatalf("health content-type = %s", h["Content-Type"])
1078 return
1079 }
1080
1081 ms, _, _ := s.routeHTTP(1, "GET", "/metrics", map[string]string{}, nil)
1082 if ms != 200 {
1083 t.Fatalf("metrics status = %d", ms)
1084 return
1085 }
1086
1087 rs, _, rb := s.routeHTTP(1, "GET", "/metrics/reset", map[string]string{}, nil)
1088 if rs != 200 || string(rb) != "reset\n" {
1089 t.Fatalf("metrics reset = %d %s", rs, rb)
1090 return
1091 }
1092
1093 ns, nh, nb := s.routeHTTP(1, "GET", "/anything", map[string]string{"accept": "application/nostr+json"}, nil)
1094 if ns != 200 || nh["Content-Type"] != "application/nostr+json" {
1095 t.Fatalf("nip11 = %d %s", ns, nh["Content-Type"])
1096 return
1097 }
1098 if !stHas(nb, []byte("\"software\"")) {
1099 t.Fatal("nip11 body must carry the software field")
1100 return
1101 }
1102
1103 fs, fh, fb := s.routeHTTP(1, "GET", "/missing", map[string]string{}, nil)
1104 if fs != 404 || string(fb) != "404 page not found\n" {
1105 t.Fatalf("fallback-less 404 = %d %s", fs, fb)
1106 return
1107 }
1108 _ = fh
1109
1110 s.Fallback = stFallback
1111 bs, bh, bb := s.routeHTTP(1, "GET", "/missing", map[string]string{}, nil)
1112 if bs != 201 || string(bb) != "fb" || bh["X-Fallback"] != "1" {
1113 t.Fatalf("fallback = %d %s %s", bs, bb, bh["X-Fallback"])
1114 return
1115 }
1116
1117 // CORS is merged into a fallback response when enabled.
1118 s.cfg.CORSEnabled = true
1119 bs2, bh2, _ := s.routeHTTP(1, "GET", "/missing", map[string]string{"origin": "https://a"}, nil)
1120 if bs2 != 201 {
1121 t.Fatalf("fallback with CORS = %d", bs2)
1122 return
1123 }
1124 if bh2["Access-Control-Allow-Origin"] != "https://a" {
1125 t.Fatal("fallback CORS headers must be merged")
1126 return
1127 }
1128 }
1129
1130 func TestOnHTTP(t *testing.T) {
1131 c := stCfg()
1132 c.CORSEnabled = true
1133 s := stServer(c)
1134 s.Version = "9"
1135
1136 status, _, body, connClose := s.OnHTTP(1, "GET", "/health", map[string]string{}, nil)
1137 if status != 200 || string(body) != "ok" || connClose {
1138 t.Fatalf("plain health = %d %s close=%v", status, body, connClose)
1139 return
1140 }
1141
1142 os_, oh, _, _ := s.OnHTTP(1, "OPTIONS", "/x", map[string]string{"origin": "https://a"}, nil)
1143 if os_ != 204 {
1144 t.Fatalf("OPTIONS = %d, want 204", os_)
1145 return
1146 }
1147 if oh["Access-Control-Allow-Origin"] != "https://a" {
1148 t.Fatal("OPTIONS must answer with CORS headers")
1149 return
1150 }
1151
1152 _, _, _, cc := s.OnHTTP(1, "GET", "/health", map[string]string{"connection": "close"}, nil)
1153 if !cc {
1154 t.Fatal("connection: close must be reported")
1155 return
1156 }
1157
1158 // Blacklist re-check. The transport has no connection for fd 1, so ConnIP
1159 // returns "", and an empty prefix matches it. That exercises the 403 branch.
1160 c.IPBlacklist = []string{""}
1161 bs, _, _, _ := s.OnHTTP(1, "GET", "/health", map[string]string{}, nil)
1162 if bs != 403 {
1163 t.Fatalf("blacklisted request = %d, want 403", bs)
1164 return
1165 }
1166 }
1167
1168 func TestOnHTTPProxyAndBlossomDispatch(t *testing.T) {
1169 ps := stProxySrv(1)
1170 ps.selfHost = "self.example/"
1171 status, _, _, _ := ps.OnHTTP(1, "GET", "/proxy/https://other.example/x", map[string]string{}, nil)
1172 if status != transport.HTTPDeferred {
1173 t.Fatalf("proxy dispatch status = %d, want deferred", status)
1174 return
1175 }
1176 if len(ps.asyncPending) != 1 || !ps.proxyPool.Busy[0] {
1177 t.Fatal("proxy dispatch must register async state")
1178 return
1179 }
1180
1181 bs := stBlossomSrv(1)
1182 bstatus, _, _, _ := bs.OnHTTP(1, "GET", "/blossom/abc", map[string]string{}, nil)
1183 if bstatus != transport.HTTPDeferred {
1184 t.Fatalf("blossom dispatch status = %d, want deferred", bstatus)
1185 return
1186 }
1187 if len(bs.asyncPending) != 1 || !bs.blossomPool.Busy[0] {
1188 t.Fatal("blossom dispatch must register async state")
1189 return
1190 }
1191 }
1192
1193 func TestDoDispatchBlossomBusy(t *testing.T) {
1194 s := stBlossomSrv(1)
1195 s.blossomPool.Busy[0] = true
1196 if !s.doDispatchBlossom(1, "GET", "/blossom/abc", map[string]string{}, nil) {
1197 t.Fatal("a busy blossom pool still reports handled")
1198 return
1199 }
1200 if len(s.asyncPending) != 0 {
1201 t.Fatal("a busy blossom pool must not register async state")
1202 return
1203 }
1204 }
1205
1206 // --- event ingestion decisions ---
1207
1208 func TestHandleEventRateLimited(t *testing.T) {
1209 c := stCfg()
1210 c.RelayURL = "wss://relay.example"
1211 s := stServer(c)
1212 s.db = stFakeDB()
1213 s.writeLimiter = ratelimit.New(1.0, 1)
1214 fd := int32(4)
1215 s.conns[fd] = &cstate{subs: map[string]*sub{}}
1216
1217 ev := stSignedEvent(t, 1, nil, "x")
1218 msg := stEventMsg(ev)
1219 s.handleEvent(fd, msg)
1220 if len(s.dbPending) != 1 {
1221 t.Fatalf("the first unauthed write must pass, dbPending = %d", len(s.dbPending))
1222 return
1223 }
1224 s.handleEvent(fd, msg)
1225 if len(s.dbPending) != 1 {
1226 t.Fatal("the rate-limited write must not reach the database")
1227 return
1228 }
1229 }
1230
1231 func TestHandleEventAuthToWrite(t *testing.T) {
1232 c := stCfg()
1233 c.RelayURL = "wss://relay.example"
1234 c.AuthToWrite = true
1235 c.FreeWriteLimit = 0 // no limiter; AuthToWrite is the only gate
1236 s := stServer(c)
1237 s.db = stFakeDB()
1238 fd := int32(4)
1239 s.conns[fd] = &cstate{subs: map[string]*sub{}}
1240
1241 ev := stSignedEvent(t, 1, nil, "x")
1242 s.handleEvent(fd, stEventMsg(ev))
1243 if len(s.dbPending) != 0 {
1244 t.Fatal("auth-required must reject an unauthed write")
1245 return
1246 }
1247 }
1248
1249 func TestHandleEventWriteExemptMarmot(t *testing.T) {
1250 // access.IsMLS reads kind package pointer globals, which stay nil until
1251 // kind.Ensure runs. handleEvent never calls Ensure, and kind state is
1252 // per-domain, so a gated relay with MarmotOpen (or NIP46BypassAuth) true
1253 // segfaults the parent domain on the first unauthed write. Reported;
1254 // Ensure here so the exemption decision itself is what gets tested.
1255 kind.Ensure()
1256 c := stCfg()
1257 c.RelayURL = "wss://relay.example"
1258 c.AuthToWrite = true
1259 c.MarmotOpen = true
1260 s := stServer(c)
1261 s.db = stFakeDB()
1262 fd := int32(4)
1263 s.conns[fd] = &cstate{subs: map[string]*sub{}}
1264
1265 ev := stSignedEvent(t, 443, nil, "mls")
1266 s.handleEvent(fd, stEventMsg(ev))
1267 if len(s.dbPending) != 1 {
1268 t.Fatalf("an exempt MLS kind must bypass auth, dbPending = %d", len(s.dbPending))
1269 return
1270 }
1271 }
1272
1273 func TestHandleEventWriteExemptNIP46(t *testing.T) {
1274 kind.Ensure()
1275 c := stCfg()
1276 c.RelayURL = "wss://relay.example"
1277 c.AuthToWrite = true
1278 c.NIP46BypassAuth = true
1279 s := stServer(c)
1280 s.db = stFakeDB()
1281 fd := int32(4)
1282 s.conns[fd] = &cstate{subs: map[string]*sub{}}
1283
1284 ev := stSignedEvent(t, 24133, nil, "connect")
1285 s.handleEvent(fd, stEventMsg(ev))
1286 if len(s.dbPending) != 1 {
1287 t.Fatalf("an exempt NIP-46 connect must bypass auth, dbPending = %d", len(s.dbPending))
1288 return
1289 }
1290 }
1291
1292 func TestHandleEventDispatchesToWorker(t *testing.T) {
1293 s := stIngestSrv(1)
1294 fd := int32(4)
1295 s.conns[fd] = &cstate{subs: map[string]*sub{}}
1296 ev := stSignedEvent(t, 1, nil, "x")
1297 s.handleEvent(fd, stEventMsg(ev))
1298 if len(s.pendingReq) != 1 || !s.workers.Busy[0] {
1299 t.Fatal("a live ingest pool must take the event")
1300 return
1301 }
1302 if len(s.dbPending) != 0 {
1303 t.Fatal("the worker path must not talk to the database directly")
1304 return
1305 }
1306 }
1307
1308 func TestHandleEventMalformedAndMissingConn(t *testing.T) {
1309 s := stServer(stCfg())
1310 s.db = stFakeDB()
1311 s.handleEvent(99, []byte("[\"EVENT\",{}]"))
1312 if len(s.dbPending) != 0 {
1313 t.Fatal("a missing connection must be a no-op")
1314 return
1315 }
1316 fd := int32(4)
1317 s.conns[fd] = &cstate{subs: map[string]*sub{}}
1318 s.handleEvent(fd, []byte("[\"EVENT\",]"))
1319 if len(s.dbPending) != 0 {
1320 t.Fatal("a malformed EVENT must not reach the database")
1321 return
1322 }
1323 }
1324
1325 // --- dispatch routing ---
1326
1327 func TestDispatchEventPersists(t *testing.T) {
1328 s := stServer(stCfg())
1329 s.db = stFakeDB()
1330 fd := int32(4)
1331 s.conns[fd] = &cstate{subs: map[string]*sub{}}
1332 ev := stSignedEvent(t, 1, nil, "hello")
1333 s.dispatch(fd, stEventMsg(ev))
1334 if len(s.dbPending) != 1 {
1335 t.Fatalf("dispatch must persist an EVENT, dbPending = %d", len(s.dbPending))
1336 return
1337 }
1338 req, ok := stTryDBReq(s.db.in)
1339 if !ok || req.Op != tree.OpPersist || req.ConnID != fd {
1340 t.Fatalf("persist request op=%d conn=%d ok=%v", req.Op, req.ConnID, ok)
1341 return
1342 }
1343 }
1344
1345 func TestDispatchJailedAndUnknown(t *testing.T) {
1346 s := stServer(stCfg())
1347 s.db = stFakeDB()
1348 fd := int32(4)
1349 cs := &cstate{subs: map[string]*sub{}, jailed: true}
1350 s.conns[fd] = cs
1351 ev := stSignedEvent(t, 1, nil, "hello")
1352 s.dispatch(fd, stEventMsg(ev))
1353 if len(s.dbPending) != 0 {
1354 t.Fatal("a jailed connection must be silent")
1355 return
1356 }
1357 s.dispatch(99, stEventMsg(ev))
1358 if len(s.dbPending) != 0 {
1359 t.Fatal("an unknown connection must be a no-op")
1360 return
1361 }
1362 // An unrecognised label is dropped by dispatch.
1363 s.conns[fd].jailed = false
1364 s.dispatch(fd, []byte("[\"NOTICE\",\"hi\"]"))
1365 if len(s.dbPending) != 0 {
1366 t.Fatal("an unknown label must be dropped")
1367 return
1368 }
1369 }
1370
1371 func TestHandleCountAndTaint(t *testing.T) {
1372 s := stServer(stCfg())
1373 s.db = stFakeDB()
1374 fd := int32(4)
1375 s.conns[fd] = &cstate{subs: map[string]*sub{}}
1376 s.handleCount(fd, []byte("[\"COUNT\",\"c1\",{}]"))
1377 if len(s.dbPending) != 1 {
1378 t.Fatalf("handleCount dbPending = %d, want 1", len(s.dbPending))
1379 return
1380 }
1381 req, ok := stTryDBReq(s.db.in)
1382 if !ok || req.Op != tree.OpCount || req.ConnID != fd {
1383 t.Fatalf("count request op=%d conn=%d ok=%v", req.Op, req.ConnID, ok)
1384 return
1385 }
1386
1387 // A malformed COUNT is dropped.
1388 s.handleCount(fd, []byte("[\"COUNT\",\"c1\"]"))
1389
1390 // An undersized hex id taints the filter and jails the connection.
1391 s.handleCount(fd, []byte("[\"COUNT\",\"c2\",{\"ids\":[\"abcd\"]}]"))
1392 if !s.conns[fd].jailed {
1393 t.Fatal("an undersized hex id must jail the connection")
1394 return
1395 }
1396
1397 // Missing connection.
1398 s.handleCount(99, []byte("[\"COUNT\",\"c1\",{}]"))
1399
1400 // Auth required and unauthed.
1401 c2 := stCfg()
1402 c2.AuthRequired = true
1403 s2 := stServer(c2)
1404 s2.db = stFakeDB()
1405 s2.conns[4] = &cstate{subs: map[string]*sub{}}
1406 s2.handleCount(4, []byte("[\"COUNT\",\"c1\",{}]"))
1407 if len(s2.dbPending) != 0 {
1408 t.Fatal("auth-required COUNT must be silent when unauthed")
1409 return
1410 }
1411 }
1412
1413 func TestHandleEventRejectReason(t *testing.T) {
1414 s := stServer(stCfg())
1415 fd := int32(4)
1416 s.conns[fd] = &cstate{subs: map[string]*sub{}}
1417 ev := stSignedEvent(t, 1, nil, "x")
1418 // Directly drive eventReject; with no transport connection the send is a
1419 // no-op, so this only proves it parses and does not panic.
1420 s.eventReject(fd, stEventMsg(ev), "rate-limited: too many events")
1421 }
1422
1423 func TestOnTickReaps(t *testing.T) {
1424 s := stServer(stCfg())
1425 now := time.Now().UnixNano()
1426 s.asyncPending[1] = asyncHTTPEntry{connFD: 3, createdAt: now - 120_000_000_000}
1427 s.OnTick()
1428 if _, ok := s.asyncPending[1]; ok {
1429 t.Fatal("OnTick must run the async reaper")
1430 return
1431 }
1432 }
1433