Coracle/nostrlib
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

boltdb: it works!

Browse Source
This commit is contained in:
fiatjaf
2025-08-05 16:25:55 -03:00
parent 144f4a4740
commit 47ca205e9e
13 changed files with 104 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
eventstore/boltdb/count.go
View File

@@ -1,4 +1,4 @@
package bolt
package boltdb
import (
"bytes"
@@ -28,13 +28,12 @@ func (b *BoltBackend) CountEvents(filter nostr.Filter) (uint32, error) {
for _, q := range queries {
cursor := txn.Bucket(q.bucket).Cursor()
it := &iterator{cursor: cursor}
it := newIterator(q, cursor)
it.seek(q.startingPoint)
for {
// we already have a k and a v and an err from the cursor setup, so check and use these
if it.err != nil ||
!bytes.HasPrefix(it.key, q.prefix) {
if !bytes.HasPrefix(it.key, q.prefix) {
// either iteration has errored or we reached the end of this prefix
break // stop this cursor and move to the next one
}
@@ -113,13 +112,12 @@ func (b *BoltBackend) CountEventsHLL(filter nostr.Filter, offset int) (uint32, *
for _, q := range queries {
cursor := txn.Bucket(q.bucket).Cursor()
it := &iterator{cursor: cursor}
it := newIterator(q, cursor)
it.seek(q.startingPoint)
for {
// we already have a k and a v and an err from the cursor setup, so check and use these
if it.err != nil ||
!bytes.HasPrefix(it.key, q.prefix) {
if !bytes.HasPrefix(it.key, q.prefix) {
// either iteration has errored or we reached the end of this prefix
break // stop this cursor and move to the next one
}

4
eventstore/boltdb/delete.go
View File

@@ -1,4 +1,4 @@
package bolt
package boltdb
import (
"fmt"
@@ -31,7 +31,7 @@ func (b *BoltBackend) delete(txn *bbolt.Tx, id nostr.ID) error {
// calculate all index keys we have for this event and delete them
for k := range b.getIndexKeysForEvent(evt) {
err := txn.Bucket(k.bucket).Delete(k.key)
err := txn.Bucket(k.bucket).Delete(k.fullkey)
if err != nil {
return fmt.Errorf("failed to delete index entry %s for %x: %w", b.keyName(k), evt.ID[0:8], err)
}

2
eventstore/boltdb/fuzz_test.go
View File

@@ -1,4 +1,4 @@
package bolt
package boltdb
import (
"cmp"

77
eventstore/boltdb/helpers.go
View File

@@ -1,4 +1,4 @@
package bolt
package boltdb
import (
"bytes"
@@ -22,7 +22,6 @@ type iterator struct {
cursor *bbolt.Cursor
key []byte
currIdPtr []byte
err error
// this keeps track of last timestamp value pulled from this
last uint32
@@ -35,16 +34,21 @@ type iterator struct {
timestamps []uint32
}
func newIterator(query query, cursor *bbolt.Cursor) *iterator {
return &iterator{
query: query,
cursor: cursor,
key: make([]byte, 0, 31),
currIdPtr: make([]byte, 8),
}
}
func (it *iterator) pull(n int, since uint32) {
query := it.query
for range n {
// in the beginning we already have a k and a v and an err from the cursor setup, so check and use these
if it.err != nil {
it.exhausted = true
return
}
// in the beginning we already have a k and a v from the cursor setup, so check and use these
if !bytes.HasPrefix(it.key, query.prefix) {
// we reached the end of this prefix
it.exhausted = true
@@ -58,29 +62,49 @@ func (it *iterator) pull(n int, since uint32) {
}
// got a key
it.idPtrs = append(it.idPtrs, it.currIdPtr)
it.idPtrs = append(it.idPtrs, append([]byte{}, it.currIdPtr...))
it.timestamps = append(it.timestamps, createdAt)
it.last = createdAt
// advance the cursor for the next call
it.next()
if it.exhausted {
return
}
}
return
}
func (it *iterator) seek(keyPrefix []byte) {
fullkey, _ := it.cursor.Seek(keyPrefix)
copy(it.key, fullkey[len(fullkey)-8-4:])
copy(it.currIdPtr, fullkey[len(fullkey)-8:])
func (it *iterator) seek(key []byte) {
fullkey, _ := it.cursor.Seek(key)
if fullkey == nil || bytes.Compare(fullkey, key) == 1 {
fullkey, _ = it.cursor.Prev()
if fullkey == nil {
it.exhausted = true
return
}
}
s := len(fullkey)
it.key = it.key[0 : s-8]
copy(it.key, fullkey[0:s-8])
copy(it.currIdPtr, fullkey[s-8:])
}
// goes backwards
func (it *iterator) next() {
// move one back (we'll look into k and v and err in the next iteration)
// move one back (we'll look into key in the next iteration)
fullkey, _ := it.cursor.Prev()
copy(it.key, fullkey[len(fullkey)-8-4:])
copy(it.currIdPtr, fullkey[len(fullkey)-8:])
if fullkey == nil {
it.exhausted = true
return
}
s := len(fullkey)
it.key = it.key[0 : s-8]
copy(it.key, fullkey[0:s-8])
copy(it.currIdPtr, fullkey[s-8:])
}
type iterators []*iterator
@@ -160,12 +184,14 @@ func (its iterators) quickselect(k int) uint32 {
}
type key struct {
bucket []byte
key []byte
bucket []byte
fullkey []byte
}
func (b *BoltBackend) keyName(key key) string {
return fmt.Sprintf("<dbi=%s key=%x>", string(key.bucket), key.key)
s := len(key.fullkey)
return fmt.Sprintf("<dbi=%s prefix=%x ts=%x idptr=%x>",
string(key.bucket), key.fullkey[0:s-8-4], key.fullkey[s-8-4:s-8], key.fullkey[s-8:])
}
func (b *BoltBackend) getIndexKeysForEvent(evt nostr.Event) iter.Seq[key] {
@@ -176,7 +202,7 @@ func (b *BoltBackend) getIndexKeysForEvent(evt nostr.Event) iter.Seq[key] {
copy(k[0:8], evt.PubKey[0:8])
binary.BigEndian.PutUint32(k[8:8+4], uint32(evt.CreatedAt))
copy(k[8+4:8+4+8], evt.ID[16:24])
if !yield(key{bucket: indexPubkey, key: k[0 : 8+4]}) {
if !yield(key{bucket: indexPubkey, fullkey: k}) {
return
}
}
@@ -187,7 +213,7 @@ func (b *BoltBackend) getIndexKeysForEvent(evt nostr.Event) iter.Seq[key] {
binary.BigEndian.PutUint16(k[0:2], uint16(evt.Kind))
binary.BigEndian.PutUint32(k[2:2+4], uint32(evt.CreatedAt))
copy(k[2+4:2+4+8], evt.ID[16:24])
if !yield(key{bucket: indexKind, key: k[0 : 2+4]}) {
if !yield(key{bucket: indexKind, fullkey: k}) {
return
}
}
@@ -198,8 +224,8 @@ func (b *BoltBackend) getIndexKeysForEvent(evt nostr.Event) iter.Seq[key] {
copy(k[0:8], evt.PubKey[0:8])
binary.BigEndian.PutUint16(k[8:8+2], uint16(evt.Kind))
binary.BigEndian.PutUint32(k[8+2:8+2+4], uint32(evt.CreatedAt))
copy(k[8+2:8+2+8], evt.ID[16:24])
if !yield(key{bucket: indexPubkeyKind, key: k[0 : 8+2+4]}) {
copy(k[8+2+4:8+2+4+8], evt.ID[16:24])
if !yield(key{bucket: indexPubkeyKind, fullkey: k}) {
return
}
}
@@ -222,9 +248,10 @@ func (b *BoltBackend) getIndexKeysForEvent(evt nostr.Event) iter.Seq[key] {
// get key prefix (with full length) and offset where to write the created_at
bucket, k := b.getTagIndexPrefix(tag[0], tag[1])
// keys always end with 4 bytes of created_at + 8 bytes of the id ptr
binary.BigEndian.PutUint32(k[len(k)-8-4:], uint32(evt.CreatedAt))
copy(k[len(k)-8:], evt.ID[16:24])
if !yield(key{bucket: bucket, key: k}) {
if !yield(key{bucket: bucket, fullkey: k}) {
return
}
}
@@ -234,7 +261,7 @@ func (b *BoltBackend) getIndexKeysForEvent(evt nostr.Event) iter.Seq[key] {
k := make([]byte, 4+8)
binary.BigEndian.PutUint32(k[0:4], uint32(evt.CreatedAt))
copy(k[4:4+8], evt.ID[16:24])
if !yield(key{bucket: indexCreatedAt, key: k[0:4]}) {
if !yield(key{bucket: indexCreatedAt, fullkey: k}) {
return
}
}

32
eventstore/boltdb/lib.go
View File

@@ -1,7 +1,7 @@
package bolt
package boltdb
import (
"os"
"time"
"fiatjaf.com/nostr"
"fiatjaf.com/nostr/eventstore"
@@ -32,24 +32,20 @@ type BoltBackend struct {
}
func (b *BoltBackend) Init() error {
// create directory if it doesn't exist and open it
if err := os.MkdirAll(b.Path, 0755); err != nil {
return err
}
return b.initialize()
}
func (b *BoltBackend) Close() {
b.DB.Close()
}
func (b *BoltBackend) initialize() error {
db, err := bbolt.Open(b.Path, 0600, nil)
db, err := bbolt.Open(b.Path, 0600, &bbolt.Options{
Timeout: 2 * time.Second,
PreLoadFreelist: true,
FreelistType: bbolt.FreelistMapType,
})
if err != nil {
return err
}
db.AllocSize = 64 * 1024 * 1024
db.MaxBatchDelay = time.Millisecond * 40
b.DB = db
db.Update(func(txn *bbolt.Tx) error {
if _, err := txn.CreateBucketIfNotExists(settingsStore); err != nil {
return err
@@ -86,3 +82,7 @@ func (b *BoltBackend) initialize() error {
return b.migrate()
}
func (b *BoltBackend) Close() {
b.DB.Close()
}

2
eventstore/boltdb/migration.go
View File

@@ -1,4 +1,4 @@
package bolt
package boltdb
import (
"encoding/binary"

21
eventstore/boltdb/query.go
View File

@@ -1,4 +1,4 @@
package bolt
package boltdb
import (
"iter"
@@ -76,20 +76,23 @@ func (b *BoltBackend) query(txn *bbolt.Tx, filter nostr.Filter, limit int, yield
return err
}
iterators := make(iterators, len(queries))
batchSizePerQuery := internal.BatchSizePerNumberOfQueries(limit, len(queries))
iterators := make(iterators, 0, len(queries))
for _, query := range queries {
bucket := txn.Bucket(query.bucket)
for q, query := range queries {
bucket := txn.Bucket(queries[q].bucket)
it := newIterator(query, bucket.Cursor())
iterators[q] = &iterator{
query: query,
cursor: bucket.Cursor(),
it.seek(query.startingPoint)
if it.exhausted {
// this may happen rarely
continue
}
iterators[q].seek(queries[q].startingPoint)
iterators = append(iterators, it)
}
batchSizePerQuery := internal.BatchSizePerNumberOfQueries(limit, len(queries))
// initial pull from all queries
for i := range iterators {
iterators[i].pull(batchSizePerQuery, since)

2
eventstore/boltdb/query_planner.go
View File

@@ -1,4 +1,4 @@
package bolt
package boltdb
import (
"encoding/binary"

2
eventstore/boltdb/replace.go
View File

@@ -1,4 +1,4 @@
package bolt
package boltdb
import (
"fmt"

4
eventstore/boltdb/save.go
View File

@@ -1,4 +1,4 @@
package bolt
package boltdb
import (
"fmt"
@@ -55,7 +55,7 @@ func (b *BoltBackend) save(txn *bbolt.Tx, evt nostr.Event) error {
// put indexes
for k := range b.getIndexKeysForEvent(evt) {
err := txn.Bucket(k.bucket).Put(k.key, nil)
err := txn.Bucket(k.bucket).Put(k.fullkey, nil)
if err != nil {
return err
}

2
eventstore/boltdb/utils_test.go
View File

@@ -1,4 +1,4 @@
package bolt
package boltdb
import (
"testing"