nostrlib/eventstore/mmm/save.go

package mmm

import (
	"encoding/binary"
	"fmt"
	"os"
	"runtime"
	"slices"
	"syscall"
	"unsafe"

	"fiatjaf.com/nostr"
	"fiatjaf.com/nostr/eventstore"
	"fiatjaf.com/nostr/eventstore/codec/betterbinary"
	"github.com/PowerDNS/lmdb-go/lmdb"
)

func (il *IndexingLayer) SaveEvent(evt nostr.Event) error {
	if il.mmmm.ReadOnly {
		return ReadOnly
	}

	il.mmmm.writeMutex.Lock()
	defer il.mmmm.writeMutex.Unlock()

	runtime.LockOSThread()
	defer runtime.UnlockOSThread()

	// prepare transactions
	mmmtxn, err := il.mmmm.lmdbEnv.BeginTxn(nil, 0)
	if err != nil {
		return err
	}
	defer func() {
		// defer abort but only if we haven't committed (we'll set it to nil after committing)
		if mmmtxn != nil {
			mmmtxn.Abort()
		}
	}()
	mmmtxn.RawRead = true

	iltxn, err := il.lmdbEnv.BeginTxn(nil, 0)
	if err != nil {
		return err
	}
	defer func() {
		// defer abort but only if we haven't committed (we'll set it to nil after committing)
		if iltxn != nil {
			iltxn.Abort()
		}
	}()
	iltxn.RawRead = true

	// the actual save operation
	if _, err := il.mmmm.storeOn(mmmtxn, iltxn, il, evt); err != nil {
		return err
	}

	// commit in this order to minimize problematic inconsistencies
	if err := mmmtxn.Commit(); err != nil {
		return fmt.Errorf("can't commit mmmtxn: %w", err)
	}
	mmmtxn = nil
	if err := iltxn.Commit(); err != nil {
		return fmt.Errorf("can't commit iltxn: %w", err)
	}
	iltxn = nil

	return nil
}

func (b *MultiMmapManager) storeOn(
	mmmtxn *lmdb.Txn,
	iltxn *lmdb.Txn,
	il *IndexingLayer,
	evt nostr.Event,
) (stored bool, err error) {
	// check if we already have this id
	var pos position
	val, err := mmmtxn.Get(b.indexId, evt.ID[0:8])
	if err == nil {
		pos = positionFromBytes(val[0:12])
		// we found the event, now check if it is already indexed by the layer that wants to store it
		for s := 12; s < len(val); s += 2 {
			ilid := binary.BigEndian.Uint16(val[s : s+2])
			if il.id == ilid {
				// already on the specified layer, we can end here
				return false, eventstore.ErrDupEvent
			}
		}
	} else if !lmdb.IsNotFound(err) {
		// if we got an error from lmdb we will only proceed if it's NotFound -- for anything else we will error
		return false, fmt.Errorf("error checking existence: %w", err)
	}

	// ok, now we have to write the event to the mmapped file
	// unless we already have the event stored, in that case we don't have to write it again, we'll just reuse it
	if val == nil {
		// get event binary size
		pos = position{
			size: uint32(betterbinary.Measure(evt)),
		}
		if pos.size >= 1<<16 {
			return false, fmt.Errorf("event too large to store, max %d, got %d", 1<<16, pos.size)
		}

		// find a suitable place for this to be stored in
		appendToMmap := true
		for f, fr := range b.freeRanges {
			if fr.size >= pos.size {
				// found the smallest possible place that can fit this event
				appendToMmap = false
				pos.start = fr.start

				// modify the free ranges we're keeping track of
				// (in case of conflict we lose this free range but it's ok, it will be recovered on the next startup)
				if pos.size == fr.size {
					// if we've used it entirely just delete it
					b.freeRanges = slices.Delete(b.freeRanges, f, f+1)
				} else {
					// otherwise modify it in place
					b.freeRanges[f] = position{
						start: fr.start + uint64(pos.size),
						size:  fr.size - pos.size,
					}
				}

				break
			}
		}

		if appendToMmap {
			// no free ranges found, so write to the end of the mmap file
			pos.start = b.mmapfEnd
			mmapfNewSize := int64(b.mmapfEnd) + int64(pos.size)
			if err := os.Truncate(b.mmapfPath, mmapfNewSize); err != nil {
				return false, fmt.Errorf("error increasing %s: %w", b.mmapfPath, err)
			}
			b.mmapfEnd = uint64(mmapfNewSize)
		}

		// write to the mmap
		if err := betterbinary.Marshal(evt, b.mmapf[pos.start:]); err != nil {
			return false, fmt.Errorf("error marshaling to %d: %w", pos.start, err)
		}

		// msync
		_, _, errno := syscall.Syscall(syscall.SYS_MSYNC,
			uintptr(unsafe.Pointer(&b.mmapf[0])), uintptr(len(b.mmapf)), syscall.MS_SYNC)
		if errno != 0 {
			panic(fmt.Errorf("msync failed: %w", syscall.Errno(errno)))
		}

		// prepare value to be saved in the id index (if we didn't have it already)
		// val: [posb][layerIdRefs...]
		val = make([]byte, 12, 12+2) // only reserve room for one layer after the position
		writeBytesFromPosition(val, pos)
	}

	// generate and save indexes
	for k := range il.getIndexKeysForEvent(evt) {
		if err := iltxn.Put(k.dbi, k.key, val[0:12] /* pos */, 0); err != nil {
			b.Logger.Warn().Str("name", il.name).Msg("failed to index event on layer")
		}
	}

	// add layer to the id index val
	val = binary.BigEndian.AppendUint16(val, il.id)

	// store the id index with the new layer reference
	if err := mmmtxn.Put(b.indexId, evt.ID[0:8], val, 0); err != nil {
		return false, fmt.Errorf("failed to store %x by id: %w", evt.ID[:], err)
	}

	return true, nil
}