mleku/next.orly.dev
1
0
Fork 1
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

Add detailed benchmark results for multiple relays.

Browse Source 
- Included results for `relayer-basic`, `strfry`, and `nostr-rs-relay` relay benchmarks.
- Comprehensive performance metrics added for throughput, latency, query, and concurrent operations.
- Reports saved as plain text and AsciiDoc formats.
This commit is contained in:
mleku
2025-09-20 12:06:57 +01:00
parent 81a40c04e5
commit 90f21fbcd1
14 changed files with 2053 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
pkg/database/fetch-event-by-serial.go
View File

@@ -2,6 +2,7 @@ package database
import (
"bytes"
"fmt"
"github.com/dgraph-io/badger/v4"
"lol.mleku.dev/chk"
@@ -25,8 +26,23 @@ func (d *D) FetchEventBySerial(ser *types.Uint40) (ev *event.E, err error) {
if v, err = item.ValueCopy(nil); chk.E(err) {
return
}
// Check if we have valid data before attempting to unmarshal
if len(v) < 32+32+1+2+1+1+64 { // ID + Pubkey + min varint fields + Sig
err = fmt.Errorf(
"incomplete event data: got %d bytes, expected at least %d",
len(v), 32+32+1+2+1+1+64,
)
return
}
ev = new(event.E)
if err = ev.UnmarshalBinary(bytes.NewBuffer(v)); chk.E(err) {
if err = ev.UnmarshalBinary(bytes.NewBuffer(v)); err != nil {
// Add more context to EOF errors for debugging
if err.Error() == "EOF" {
err = fmt.Errorf(
"EOF while unmarshaling event (serial=%v, data_len=%d): %w",
ser, len(v), err,
)
}
return
}
return

124
pkg/database/fetch-events-by-serials.go
View File

@@ -2,105 +2,67 @@ package database
import (
"bytes"
"sort"
"strconv"
"github.com/dgraph-io/badger/v4"
"lol.mleku.dev/chk"
"lol.mleku.dev/log"
"next.orly.dev/pkg/database/indexes"
"next.orly.dev/pkg/database/indexes/types"
"next.orly.dev/pkg/encoders/event"
)
// FetchEventsBySerials processes multiple serials in ascending order and retrieves
// the corresponding events from the database. It optimizes database access by
// sorting the serials and seeking to each one sequentially.
func (d *D) FetchEventsBySerials(serials []*types.Uint40) (
evMap map[string]*event.E, err error,
) {
log.T.F("FetchEventsBySerials: processing %d serials", len(serials))
// Initialize the result map
evMap = make(map[string]*event.E)
// Return early if no serials are provided
// FetchEventsBySerials fetches multiple events by their serials in a single database transaction.
// Returns a map of serial uint64 value to event, only including successfully fetched events.
func (d *D) FetchEventsBySerials(serials []*types.Uint40) (events map[uint64]*event.E, err error) {
events = make(map[uint64]*event.E)
if len(serials) == 0 {
return
return events, nil
}
// Sort serials in ascending order for more efficient database access
sortedSerials := make([]*types.Uint40, len(serials))
copy(sortedSerials, serials)
sort.Slice(
sortedSerials, func(i, j int) bool {
return sortedSerials[i].Get() < sortedSerials[j].Get()
},
)
// Process all serials in a single transaction
if err = d.View(
func(txn *badger.Txn) (err error) {
// Create an iterator with default options
it := txn.NewIterator(badger.DefaultIteratorOptions)
defer it.Close()
// Process each serial sequentially
for _, ser := range sortedSerials {
// Create the key for this serial
for _, ser := range serials {
buf := new(bytes.Buffer)
if err = indexes.EventEnc(ser).MarshalWrite(buf); chk.E(err) {
// Skip this serial on error but continue with others
continue
}
key := buf.Bytes()
// Seek to this key in the database
it.Seek(key)
if it.Valid() {
item := it.Item()
// Verify the key matches exactly (should always be true after a Seek)
if !bytes.Equal(item.Key(), key) {
continue
}
ev := new(event.E)
if err = item.Value(
func(val []byte) (err error) {
// Unmarshal the event
if err = ev.UnmarshalBinary(bytes.NewBuffer(val)); chk.E(err) {
return
}
// Store the event in the result map using the serial value as string key
return
},
); chk.E(err) {
continue
}
evMap[strconv.FormatUint(ser.Get(), 10)] = ev
// // Get the item value
// var v []byte
// if v, err = item.ValueCopy(nil); chk.E(err) {
// continue
// }
//
// // Unmarshal the event
// ev := new(event.E)
// if err = ev.UnmarshalBinary(bytes.NewBuffer(v)); chk.E(err) {
// continue
// }
var item *badger.Item
if item, err = txn.Get(buf.Bytes()); err != nil {
// Skip this serial if not found but continue with others
err = nil
continue
}
var v []byte
if v, err = item.ValueCopy(nil); chk.E(err) {
// Skip this serial on error but continue with others
err = nil
continue
}
// Check if we have valid data before attempting to unmarshal
if len(v) < 32+32+1+2+1+1+64 { // ID + Pubkey + min varint fields + Sig
// Skip this serial - incomplete data
continue
}
ev := new(event.E)
if err = ev.UnmarshalBinary(bytes.NewBuffer(v)); err != nil {
// Skip this serial on unmarshal error but continue with others
err = nil
continue
}
// Successfully unmarshaled event, add to results
events[ser.Get()] = ev
}
return
return nil
},
); chk.E(err) {
); err != nil {
return
}
log.T.F(
"FetchEventsBySerials: found %d events out of %d requested serials",
len(evMap), len(serials),
)
return
}
return events, nil
}

65
pkg/database/query-events.go
View File

@@ -52,15 +52,29 @@ func (d *D) QueryEvents(c context.Context, f *filter.F) (
// Continue with whatever IDs we found
}
// Process each found serial, fetch the event, and apply filters
// Convert serials map to slice for batch fetch
var serialsSlice []*types.Uint40
idHexToSerial := make(map[uint64]string) // Map serial value back to original ID hex
for idHex, ser := range serials {
// fetch the event
var ev *event.E
if ev, err = d.FetchEventBySerial(ser); err != nil {
log.T.F(
"QueryEvents: fetch by serial failed for id=%s ser=%v err=%v",
idHex, ser, err,
)
serialsSlice = append(serialsSlice, ser)
idHexToSerial[ser.Get()] = idHex
}
// Fetch all events in a single batch operation
var fetchedEvents map[uint64]*event.E
if fetchedEvents, err = d.FetchEventsBySerials(serialsSlice); err != nil {
log.E.F("QueryEvents: batch fetch failed: %v", err)
return
}
// Process each successfully fetched event and apply filters
for serialValue, ev := range fetchedEvents {
idHex := idHexToSerial[serialValue]
// Convert serial value back to Uint40 for expiration handling
ser := new(types.Uint40)
if err = ser.Set(serialValue); err != nil {
log.T.F("QueryEvents: error converting serial %d: %v", serialValue, err)
continue
}
@@ -134,16 +148,33 @@ func (d *D) QueryEvents(c context.Context, f *filter.F) (
// Add deletion events to the list of events to process
idPkTs = append(idPkTs, deletionIdPkTs...)
}
// First pass: collect all deletion events
// Prepare serials for batch fetch
var allSerials []*types.Uint40
serialToIdPk := make(map[uint64]*store.IdPkTs)
for _, idpk := range idPkTs {
var ev *event.E
ser := new(types.Uint40)
if err = ser.Set(idpk.Ser); chk.E(err) {
if err = ser.Set(idpk.Ser); err != nil {
continue
}
if ev, err = d.FetchEventBySerial(ser); err != nil {
allSerials = append(allSerials, ser)
serialToIdPk[ser.Get()] = idpk
}
// Fetch all events in batch
var allEvents map[uint64]*event.E
if allEvents, err = d.FetchEventsBySerials(allSerials); err != nil {
log.E.F("QueryEvents: batch fetch failed in non-IDs path: %v", err)
return
}
// First pass: collect all deletion events
for serialValue, ev := range allEvents {
// Convert serial value back to Uint40 for expiration handling
ser := new(types.Uint40)
if err = ser.Set(serialValue); err != nil {
continue
}
// check for an expiration tag and delete after returning the result
if CheckExpiration(ev) {
expDeletes = append(expDeletes, ser)
@@ -267,15 +298,7 @@ func (d *D) QueryEvents(c context.Context, f *filter.F) (
}
}
// Second pass: process all events, filtering out deleted ones
for _, idpk := range idPkTs {
var ev *event.E
ser := new(types.Uint40)
if err = ser.Set(idpk.Ser); chk.E(err) {
continue
}
if ev, err = d.FetchEventBySerial(ser); err != nil {
continue
}
for _, ev := range allEvents {
// Add logging for tag filter debugging
if f.Tags != nil && f.Tags.Len() > 0 {
// var eventTags []string