sane defaults: increase initial connections limit burst.

pablo's crash, see https://t.me/nip29_temp/670

deleting.go

@@ -3,6 +3,8 @@ package khatru
 import (
 	"context"
 	"fmt"
+	"strconv"
+	"strings"
 
 	"github.com/nbd-wtf/go-nostr"
 )
@@ -10,10 +12,35 @@ import (
 func (rl *Relay) handleDeleteRequest(ctx context.Context, evt *nostr.Event) error {
 	// event deletion -- nip09
 	for _, tag := range evt.Tags {
-		if len(tag) >= 2 && tag[0] == "e" {
+		if len(tag) >= 2 {
-			// first we fetch the event
+			var f nostr.Filter
+
+			switch tag[0] {
+			case "e":
+				f = nostr.Filter{IDs: []string{tag[1]}}
+			case "a":
+				spl := strings.Split(tag[1], ":")
+				if len(spl) != 3 {
+					continue
+				}
+				kind, err := strconv.Atoi(spl[0])
+				if err != nil {
+					continue
+				}
+				author := spl[1]
+				identifier := spl[2]
+				f = nostr.Filter{
+					Kinds:   []int{kind},
+					Authors: []string{author},
+					Tags:    nostr.TagMap{"d": []string{identifier}},
+					Until:   &evt.CreatedAt,
+				}
+			default:
+				continue
+			}
+
 			for _, query := range rl.QueryEvents {
-				ch, err := query(ctx, nostr.Filter{IDs: []string{tag[1]}})
+				ch, err := query(ctx, f)
 				if err != nil {
 					continue
 				}
@@ -24,13 +51,14 @@ func (rl *Relay) handleDeleteRequest(ctx context.Context, evt *nostr.Event) erro
 				// got the event, now check if the user can delete it
 				acceptDeletion := target.PubKey == evt.PubKey
 				var msg string
-				if acceptDeletion == false {
+				if !acceptDeletion {
 					msg = "you are not the author of this event"
 				}
 				// but if we have a function to overwrite this outcome, use that instead
 				for _, odo := range rl.OverwriteDeletionOutcome {
 					acceptDeletion, msg = odo(ctx, target, evt)
 				}
+
 				if acceptDeletion {
 					// delete it
 					for _, del := range rl.DeleteEvent {

docs/cookbook/auth.md

@@ -30,7 +30,7 @@ If on `RejectFilter` or `RejectEvent` you prefix the message with `auth-required
 relay.RejectFilter = append(relay.RejectFilter, func(ctx context.Context, filter nostr.Filter) (bool, string) {
 	return true, "auth-required: this query requires you to be authenticated"
 })
-relay.RejectEvent = append(relay.RejectFilter, func(ctx context.Context, event *nostr.Event) (bool, string) {
+relay.RejectEvent = append(relay.RejectEvent, func(ctx context.Context, event *nostr.Event) (bool, string) {
 	return true, "auth-required: publishing this event requires authentication"
 })
 ```

docs/cookbook/eventstore.md

@@ -44,7 +44,7 @@ func main() {
 }
 ```
 
-Other local key-value embedded databases that work the same way are [LMDB](https://pkg.go.dev/github.com/fiatjaf/eventstore/lmdb) and [BoltDB](https://pkg.go.dev/github.com/fiatjaf/eventstore/bolt).
+[LMDB](https://pkg.go.dev/github.com/fiatjaf/eventstore/lmdb) works the same way.
 
 [SQLite](https://pkg.go.dev/github.com/fiatjaf/eventstore/sqlite3) also stores things locally so it only needs a `Path`.
 

docs/index.md

@@ -39,3 +39,21 @@ features:
     link: https://pkg.go.dev/github.com/fiatjaf/khatru
     details: That means it is fast and lightweight, you can learn the language in 5 minutes and it builds your relay into a single binary that's easy to ship and deploy.
 ---
+
+## A glimpse of `khatru`'s power
+
+It allows you to create a fully-functional relay in 7 lines of code:
+
+```go
+func main() {
+	relay := khatru.NewRelay()
+	db := badger.BadgerBackend{Path: "/tmp/khatru-badgern-tmp"}
+    db.Init()
+	relay.StoreEvent = append(relay.StoreEvent, db.SaveEvent)
+	relay.QueryEvents = append(relay.QueryEvents, db.QueryEvents)
+	relay.DeleteEvent = append(relay.DeleteEvent, db.DeleteEvent)
+	http.ListenAndServe(":3334", relay)
+}
+```
+
+After that you can customize it in infinite ways. See the links above.

go.mod

@@ -1,12 +1,12 @@
 module github.com/fiatjaf/khatru
 
-go 1.21.4
+go 1.23
 
 require (
 	github.com/fasthttp/websocket v1.5.7
 	github.com/fiatjaf/eventstore v0.5.1
 	github.com/nbd-wtf/go-nostr v0.34.5
-	github.com/puzpuzpuz/xsync/v3 v3.0.2
+	github.com/puzpuzpuz/xsync/v3 v3.4.0
 	github.com/rs/cors v1.7.0
 	github.com/stretchr/testify v1.9.0
 )

go.sum

@@ -124,8 +124,8 @@ github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
-github.com/puzpuzpuz/xsync/v3 v3.0.2 h1:3yESHrRFYr6xzkz61LLkvNiPFXxJEAABanTQpKbAaew=
+github.com/puzpuzpuz/xsync/v3 v3.4.0 h1:DuVBAdXuGFHv8adVXjWWZ63pJq+NRXOWVXlKDBZ+mJ4=
-github.com/puzpuzpuz/xsync/v3 v3.0.2/go.mod h1:VjzYrABPabuM4KyBh1Ftq6u8nhwY5tBPKP9jpmh0nnA=
+github.com/puzpuzpuz/xsync/v3 v3.4.0/go.mod h1:VjzYrABPabuM4KyBh1Ftq6u8nhwY5tBPKP9jpmh0nnA=
 github.com/rs/cors v1.7.0 h1:+88SsELBHx5r+hZ8TCkggzSstaWNbDvThkVK8H6f9ik=
 github.com/rs/cors v1.7.0/go.mod h1:gFx+x8UowdsKA9AchylcLynDq+nNFfI8FkUZdN/jGCU=
 github.com/savsgio/gotils v0.0.0-20230208104028-c358bd845dee h1:8Iv5m6xEo1NR1AvpV+7XmhI4r39LGNzwUL4YpMuL5vk=

handlers.go

@@ -177,6 +177,7 @@ func (rl *Relay) HandleWebsocket(w http.ResponseWriter, r *http.Request) {
 					var ok bool
 					var writeErr error
 					var skipBroadcast bool
+
 					if env.Event.Kind == 5 {
 						// this always returns "blocked: " whenever it returns an error
 						writeErr = srl.handleDeleteRequest(ctx, &env.Event)

listener.go

@@ -87,7 +87,7 @@ func (rl *Relay) removeListenerId(ws *WebSocket, id string) {
 					// so its .index reflects its new position on srl.listeners
 					movedSpecs := rl.clients[moved.ws]
 					idx := slices.IndexFunc(movedSpecs, func(ls listenerSpec) bool {
-						return ls.index == movedFromIndex
+						return ls.index == movedFromIndex && ls.subrelay == srl
 					})
 					movedSpecs[idx].index = spec.index
 					rl.clients[moved.ws] = movedSpecs
@@ -121,7 +121,7 @@ func (rl *Relay) removeClientAndListeners(ws *WebSocket) {
 				// so its .index reflects its new position on srl.listeners
 				movedSpecs := rl.clients[moved.ws]
 				idx := slices.IndexFunc(movedSpecs, func(ls listenerSpec) bool {
-					return ls.index == movedFromIndex
+					return ls.index == movedFromIndex && ls.subrelay == srl
 				})
 				movedSpecs[idx].index = spec.index
 				rl.clients[moved.ws] = movedSpecs

listener_fuzz_test.go

@@ -0,0 +1,188 @@
+package khatru
+
+import (
+	"math/rand"
+	"testing"
+
+	"github.com/nbd-wtf/go-nostr"
+	"github.com/stretchr/testify/require"
+)
+
+func FuzzRandomListenerClientRemoving(f *testing.F) {
+	f.Add(uint(20), uint(20), uint(1))
+	f.Fuzz(func(t *testing.T, utw uint, ubs uint, ualf uint) {
+		totalWebsockets := int(utw)
+		baseSubs := int(ubs)
+		addListenerFreq := int(ualf) + 1
+
+		rl := NewRelay()
+
+		f := nostr.Filter{Kinds: []int{1}}
+		cancel := func(cause error) {}
+
+		websockets := make([]*WebSocket, 0, totalWebsockets*baseSubs)
+
+		l := 0
+
+		for i := 0; i < totalWebsockets; i++ {
+			ws := &WebSocket{}
+			websockets = append(websockets, ws)
+			rl.clients[ws] = nil
+		}
+
+		s := 0
+		for j := 0; j < baseSubs; j++ {
+			for i := 0; i < totalWebsockets; i++ {
+				ws := websockets[i]
+				w := idFromSeqUpper(i)
+
+				if s%addListenerFreq == 0 {
+					l++
+					rl.addListener(ws, w+":"+idFromSeqLower(j), rl, f, cancel)
+				}
+
+				s++
+			}
+		}
+
+		require.Len(t, rl.clients, totalWebsockets)
+		require.Len(t, rl.listeners, l)
+
+		for ws := range rl.clients {
+			rl.removeClientAndListeners(ws)
+		}
+
+		require.Len(t, rl.clients, 0)
+		require.Len(t, rl.listeners, 0)
+	})
+}
+
+func FuzzRandomListenerIdRemoving(f *testing.F) {
+	f.Add(uint(20), uint(20), uint(1), uint(4))
+	f.Fuzz(func(t *testing.T, utw uint, ubs uint, ualf uint, ualef uint) {
+		totalWebsockets := int(utw)
+		baseSubs := int(ubs)
+		addListenerFreq := int(ualf) + 1
+		addExtraListenerFreq := int(ualef) + 1
+
+		if totalWebsockets > 1024 || baseSubs > 1024 {
+			return
+		}
+
+		rl := NewRelay()
+
+		f := nostr.Filter{Kinds: []int{1}}
+		cancel := func(cause error) {}
+		websockets := make([]*WebSocket, 0, totalWebsockets)
+
+		type wsid struct {
+			ws *WebSocket
+			id string
+		}
+
+		subs := make([]wsid, 0, totalWebsockets*baseSubs)
+		extra := 0
+
+		for i := 0; i < totalWebsockets; i++ {
+			ws := &WebSocket{}
+			websockets = append(websockets, ws)
+			rl.clients[ws] = nil
+		}
+
+		s := 0
+		for j := 0; j < baseSubs; j++ {
+			for i := 0; i < totalWebsockets; i++ {
+				ws := websockets[i]
+				w := idFromSeqUpper(i)
+
+				if s%addListenerFreq == 0 {
+					id := w + ":" + idFromSeqLower(j)
+					rl.addListener(ws, id, rl, f, cancel)
+					subs = append(subs, wsid{ws, id})
+
+					if s%addExtraListenerFreq == 0 {
+						rl.addListener(ws, id, rl, f, cancel)
+						extra++
+					}
+				}
+
+				s++
+			}
+		}
+
+		require.Len(t, rl.clients, totalWebsockets)
+		require.Len(t, rl.listeners, len(subs)+extra)
+
+		rand.Shuffle(len(subs), func(i, j int) {
+			subs[i], subs[j] = subs[j], subs[i]
+		})
+		for _, wsidToRemove := range subs {
+			rl.removeListenerId(wsidToRemove.ws, wsidToRemove.id)
+		}
+
+		require.Len(t, rl.listeners, 0)
+		require.Len(t, rl.clients, totalWebsockets)
+		for _, specs := range rl.clients {
+			require.Len(t, specs, 0)
+		}
+	})
+}
+
+func FuzzRouterListenersPabloCrash(f *testing.F) {
+	f.Add(uint(3), uint(6), uint(2), uint(20))
+	f.Fuzz(func(t *testing.T, totalRelays uint, totalConns uint, subFreq uint, subIterations uint) {
+		totalRelays++
+		totalConns++
+		subFreq++
+		subIterations++
+
+		rl := NewRelay()
+
+		relays := make([]*Relay, int(totalRelays))
+		for i := 0; i < int(totalRelays); i++ {
+			relays[i] = NewRelay()
+		}
+
+		conns := make([]*WebSocket, int(totalConns))
+		for i := 0; i < int(totalConns); i++ {
+			ws := &WebSocket{}
+			conns[i] = ws
+			rl.clients[ws] = make([]listenerSpec, 0, subIterations)
+		}
+
+		f := nostr.Filter{Kinds: []int{1}}
+		cancel := func(cause error) {}
+
+		type wsid struct {
+			ws *WebSocket
+			id string
+		}
+
+		s := 0
+		subs := make([]wsid, 0, subIterations*totalConns*totalRelays)
+		for i, conn := range conns {
+			w := idFromSeqUpper(i)
+			for j := 0; j < int(subIterations); j++ {
+				id := w + ":" + idFromSeqLower(j)
+				for _, rlt := range relays {
+					if s%int(subFreq) == 0 {
+						rl.addListener(conn, id, rlt, f, cancel)
+						subs = append(subs, wsid{conn, id})
+					}
+					s++
+				}
+			}
+		}
+
+		for _, wsid := range subs {
+			rl.removeListenerId(wsid.ws, wsid.id)
+		}
+
+		for _, wsid := range subs {
+			require.Len(t, rl.clients[wsid.ws], 0)
+		}
+		for _, rlt := range relays {
+			require.Len(t, rlt.listeners, 0)
+		}
+	})
+}

listener_test.go

@@ -2,12 +2,27 @@ package khatru
 
 import (
 	"math/rand"
+	"strings"
 	"testing"
 
 	"github.com/nbd-wtf/go-nostr"
 	"github.com/stretchr/testify/require"
 )
 
+func idFromSeqUpper(seq int) string { return idFromSeq(seq, 65, 90) }
+func idFromSeqLower(seq int) string { return idFromSeq(seq, 97, 122) }
+func idFromSeq(seq int, min, max int) string {
+	maxSeq := max - min + 1
+	nLetters := seq/maxSeq + 1
+	result := strings.Builder{}
+	result.Grow(nLetters)
+	for l := 0; l < nLetters; l++ {
+		letter := rune(seq%maxSeq + min)
+		result.WriteRune(letter)
+	}
+	return result.String()
+}
+
 func TestListenerSetupAndRemoveOnce(t *testing.T) {
 	rl := NewRelay()
 
@@ -425,11 +440,11 @@ func TestRandomListenerClientRemoving(t *testing.T) {
 	for j := 0; j < 20; j++ {
 		for i := 0; i < 20; i++ {
 			ws := websockets[i]
-			w := string(rune(i + 65))
+			w := idFromSeqUpper(i)
 
 			if rand.Intn(2) < 1 {
 				l++
-				rl.addListener(ws, w+":"+string(rune(j+97)), rl, f, cancel)
+				rl.addListener(ws, w+":"+idFromSeqLower(j), rl, f, cancel)
 			}
 		}
 	}
@@ -470,10 +485,10 @@ func TestRandomListenerIdRemoving(t *testing.T) {
 	for j := 0; j < 20; j++ {
 		for i := 0; i < 20; i++ {
 			ws := websockets[i]
-			w := string(rune(i + 65))
+			w := idFromSeqUpper(i)
 
 			if rand.Intn(2) < 1 {
-				id := w + ":" + string(rune(j+97))
+				id := w + ":" + idFromSeqLower(j)
 				rl.addListener(ws, id, rl, f, cancel)
 				subs = append(subs, wsid{ws, id})
 
@@ -501,3 +516,30 @@ func TestRandomListenerIdRemoving(t *testing.T) {
 		require.Len(t, specs, 0)
 	}
 }
+
+func TestRouterListenersPabloCrash(t *testing.T) {
+	rl := NewRelay()
+
+	rla := NewRelay()
+	rlb := NewRelay()
+
+	ws1 := &WebSocket{}
+	ws2 := &WebSocket{}
+	ws3 := &WebSocket{}
+
+	rl.clients[ws1] = nil
+	rl.clients[ws2] = nil
+	rl.clients[ws3] = nil
+
+	f := nostr.Filter{Kinds: []int{1}}
+	cancel := func(cause error) {}
+
+	rl.addListener(ws1, ":1", rla, f, cancel)
+	rl.addListener(ws2, ":1", rlb, f, cancel)
+	rl.addListener(ws3, "a", rlb, f, cancel)
+	rl.addListener(ws3, "b", rla, f, cancel)
+	rl.addListener(ws3, "c", rlb, f, cancel)
+
+	rl.removeClientAndListeners(ws1)
+	rl.removeClientAndListeners(ws3)
+}

policies/events.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"slices"
 	"strings"
+	"time"
 
 	"github.com/nbd-wtf/go-nostr"
 )
@@ -79,7 +80,8 @@ func RestrictToSpecifiedKinds(kinds ...uint16) func(context.Context, *nostr.Even
 	}
 }
 
-func PreventTimestampsInThePast(thresholdSeconds nostr.Timestamp) func(context.Context, *nostr.Event) (bool, string) {
+func PreventTimestampsInThePast(threshold time.Duration) func(context.Context, *nostr.Event) (bool, string) {
+	thresholdSeconds := nostr.Timestamp(threshold.Seconds())
 	return func(ctx context.Context, event *nostr.Event) (reject bool, msg string) {
 		if nostr.Now()-event.CreatedAt > thresholdSeconds {
 			return true, "event too old"
@@ -88,7 +90,8 @@ func PreventTimestampsInThePast(thresholdSeconds nostr.Timestamp) func(context.C
 	}
 }
 
-func PreventTimestampsInTheFuture(thresholdSeconds nostr.Timestamp) func(context.Context, *nostr.Event) (bool, string) {
+func PreventTimestampsInTheFuture(threshold time.Duration) func(context.Context, *nostr.Event) (bool, string) {
+	thresholdSeconds := nostr.Timestamp(threshold.Seconds())
 	return func(ctx context.Context, event *nostr.Event) (reject bool, msg string) {
 		if event.CreatedAt-nostr.Now() > thresholdSeconds {
 			return true, "event too much in the future"

policies/sane_defaults.go

@@ -19,6 +19,6 @@ func ApplySaneDefaults(relay *khatru.Relay) {
 	)
 
 	relay.RejectConnection = append(relay.RejectConnection,
-		ConnectionRateLimiter(1, time.Minute*5, 3),
+		ConnectionRateLimiter(1, time.Minute*5, 10),
 	)
 }