Improve logging, error handling for ID queries, and ensure inclusive range boundaries in event management.
This commit is contained in:
@@ -15,13 +15,7 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
func (l *Listener) HandleMessage(msg []byte, remote string) {
|
func (l *Listener) HandleMessage(msg []byte, remote string) {
|
||||||
log.D.C(
|
log.D.F("%s received message:\n%s", remote, msg)
|
||||||
func() string {
|
|
||||||
return fmt.Sprintf(
|
|
||||||
"%s received message:\n%s", remote, msg,
|
|
||||||
)
|
|
||||||
},
|
|
||||||
)
|
|
||||||
var err error
|
var err error
|
||||||
var t string
|
var t string
|
||||||
var rem []byte
|
var rem []byte
|
||||||
|
|||||||
@@ -89,12 +89,23 @@ func GetIndexesFromFilter(f *filter.F) (idxs []Range, err error) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
buf := new(bytes.Buffer)
|
buf := new(bytes.Buffer)
|
||||||
|
// Create an index prefix without the serial number
|
||||||
idx := indexes.IdEnc(i, nil)
|
idx := indexes.IdEnc(i, nil)
|
||||||
if err = idx.MarshalWrite(buf); chk.E(err) {
|
if err = idx.MarshalWrite(buf); chk.E(err) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
b := buf.Bytes()
|
b := buf.Bytes()
|
||||||
r := Range{b, b}
|
|
||||||
|
// Create range that will match any serial value with this ID prefix
|
||||||
|
end := make([]byte, len(b))
|
||||||
|
copy(end, b)
|
||||||
|
|
||||||
|
// Fill the end range with 0xff bytes to match all possible serial values
|
||||||
|
for i := 0; i < 5; i++ {
|
||||||
|
end = append(end, 0xff)
|
||||||
|
}
|
||||||
|
|
||||||
|
r := Range{b, end}
|
||||||
idxs = append(idxs, r)
|
idxs = append(idxs, r)
|
||||||
return
|
return
|
||||||
}(); chk.E(err) {
|
}(); chk.E(err) {
|
||||||
|
|||||||
@@ -19,9 +19,16 @@ func (d *D) GetSerialById(id []byte) (ser *types.Uint40, err error) {
|
|||||||
if idxs, err = GetIndexesFromFilter(&filter.F{Ids: tag.NewFromBytesSlice(id)}); chk.E(err) {
|
if idxs, err = GetIndexesFromFilter(&filter.F{Ids: tag.NewFromBytesSlice(id)}); chk.E(err) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
for i, idx := range idxs {
|
||||||
|
log.T.F("GetSerialById: searching range %d: start=%x, end=%x", i, idx.Start, idx.End)
|
||||||
|
}
|
||||||
if len(idxs) == 0 {
|
if len(idxs) == 0 {
|
||||||
err = errorf.E("no indexes found for id %0x", id)
|
err = errorf.E("no indexes found for id %0x", id)
|
||||||
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
idFound := false
|
||||||
if err = d.View(
|
if err = d.View(
|
||||||
func(txn *badger.Txn) (err error) {
|
func(txn *badger.Txn) (err error) {
|
||||||
it := txn.NewIterator(badger.DefaultIteratorOptions)
|
it := txn.NewIterator(badger.DefaultIteratorOptions)
|
||||||
@@ -36,15 +43,22 @@ func (d *D) GetSerialById(id []byte) (ser *types.Uint40, err error) {
|
|||||||
if err = ser.UnmarshalRead(buf); chk.E(err) {
|
if err = ser.UnmarshalRead(buf); chk.E(err) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
idFound = true
|
||||||
} else {
|
} else {
|
||||||
// just don't return what we don't have? others may be
|
// Item not found in database
|
||||||
// found tho.
|
log.T.F("GetSerialById: ID not found in database: %s", hex.Enc(id))
|
||||||
}
|
}
|
||||||
return
|
return
|
||||||
},
|
},
|
||||||
); chk.E(err) {
|
); chk.E(err) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if !idFound {
|
||||||
|
err = errorf.E("id not found in database: %s", hex.Enc(id))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -20,7 +20,15 @@ func (d *D) GetSerialsByRange(idx Range) (
|
|||||||
},
|
},
|
||||||
)
|
)
|
||||||
defer it.Close()
|
defer it.Close()
|
||||||
for it.Seek(idx.End); it.Valid(); it.Next() {
|
// Start from a position that includes the end boundary (until timestamp)
|
||||||
|
// We create an end boundary that's slightly beyond the actual end to ensure inclusivity
|
||||||
|
endBoundary := make([]byte, len(idx.End))
|
||||||
|
copy(endBoundary, idx.End)
|
||||||
|
// Add 0xff bytes to ensure we capture all events at the exact until timestamp
|
||||||
|
for i := 0; i < 5; i++ {
|
||||||
|
endBoundary = append(endBoundary, 0xff)
|
||||||
|
}
|
||||||
|
for it.Seek(endBoundary); it.Valid(); it.Next() {
|
||||||
item := it.Item()
|
item := it.Item()
|
||||||
var key []byte
|
var key []byte
|
||||||
key = item.Key()
|
key = item.Key()
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"sort"
|
"sort"
|
||||||
"strconv"
|
"strconv"
|
||||||
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"crypto.orly/sha256"
|
"crypto.orly/sha256"
|
||||||
@@ -42,13 +43,28 @@ func (d *D) QueryEvents(c context.Context, f *filter.F) (
|
|||||||
var expDeletes types.Uint40s
|
var expDeletes types.Uint40s
|
||||||
var expEvs event.S
|
var expEvs event.S
|
||||||
if f.Ids != nil && f.Ids.Len() > 0 {
|
if f.Ids != nil && f.Ids.Len() > 0 {
|
||||||
|
for _, id := range f.Ids.T {
|
||||||
|
log.T.F("QueryEvents: looking for ID=%s", hex.Enc(id))
|
||||||
|
}
|
||||||
log.T.F("QueryEvents: ids path, count=%d", f.Ids.Len())
|
log.T.F("QueryEvents: ids path, count=%d", f.Ids.Len())
|
||||||
for _, idx := range f.Ids.T {
|
for _, idx := range f.Ids.T {
|
||||||
log.T.F("QueryEvents: lookup id=%s", hex.Enc(idx))
|
log.T.F("QueryEvents: lookup id=%s", hex.Enc(idx))
|
||||||
// we know there is only Ids in this, so run the ID query and fetch.
|
// we know there is only Ids in this, so run the ID query and fetch.
|
||||||
var ser *types.Uint40
|
var ser *types.Uint40
|
||||||
if ser, err = d.GetSerialById(idx); chk.E(err) {
|
var idErr error
|
||||||
log.T.F("QueryEvents: id miss or error id=%s err=%v", hex.Enc(idx), err)
|
if ser, idErr = d.GetSerialById(idx); idErr != nil {
|
||||||
|
// Check if this is a "not found" error which is expected for IDs we don't have
|
||||||
|
if strings.Contains(idErr.Error(), "id not found in database") {
|
||||||
|
log.T.F("QueryEvents: ID not found in database: %s", hex.Enc(idx))
|
||||||
|
} else {
|
||||||
|
// Log unexpected errors but continue processing other IDs
|
||||||
|
log.E.F("QueryEvents: error looking up id=%s err=%v", hex.Enc(idx), idErr)
|
||||||
|
}
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
// Check if the serial is nil, which indicates the ID wasn't found
|
||||||
|
if ser == nil {
|
||||||
|
log.T.F("QueryEvents: Serial is nil for ID: %s", hex.Enc(idx))
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
// fetch the events
|
// fetch the events
|
||||||
@@ -60,6 +76,7 @@ func (d *D) QueryEvents(c context.Context, f *filter.F) (
|
|||||||
log.T.F("QueryEvents: found id=%s kind=%d created_at=%d", hex.Enc(ev.ID), ev.Kind, ev.CreatedAt)
|
log.T.F("QueryEvents: found id=%s kind=%d created_at=%d", hex.Enc(ev.ID), ev.Kind, ev.CreatedAt)
|
||||||
// check for an expiration tag and delete after returning the result
|
// check for an expiration tag and delete after returning the result
|
||||||
if CheckExpiration(ev) {
|
if CheckExpiration(ev) {
|
||||||
|
log.T.F("QueryEvents: id=%s filtered out due to expiration", hex.Enc(ev.ID))
|
||||||
expDeletes = append(expDeletes, ser)
|
expDeletes = append(expDeletes, ser)
|
||||||
expEvs = append(expEvs, ev)
|
expEvs = append(expEvs, ev)
|
||||||
continue
|
continue
|
||||||
@@ -69,6 +86,7 @@ func (d *D) QueryEvents(c context.Context, f *filter.F) (
|
|||||||
log.T.F("QueryEvents: id=%s filtered out due to deletion: %v", hex.Enc(ev.ID), derr)
|
log.T.F("QueryEvents: id=%s filtered out due to deletion: %v", hex.Enc(ev.ID), derr)
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
log.T.F("QueryEvents: id=%s SUCCESSFULLY FOUND, adding to results", hex.Enc(ev.ID))
|
||||||
evs = append(evs, ev)
|
evs = append(evs, ev)
|
||||||
}
|
}
|
||||||
// sort the events by timestamp
|
// sort the events by timestamp
|
||||||
|
|||||||
@@ -3,11 +3,13 @@ package database
|
|||||||
import (
|
import (
|
||||||
"bytes"
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
|
"strings"
|
||||||
|
|
||||||
"database.orly/indexes"
|
"database.orly/indexes"
|
||||||
"database.orly/indexes/types"
|
"database.orly/indexes/types"
|
||||||
"encoders.orly/event"
|
"encoders.orly/event"
|
||||||
"encoders.orly/filter"
|
"encoders.orly/filter"
|
||||||
|
"encoders.orly/hex"
|
||||||
"encoders.orly/kind"
|
"encoders.orly/kind"
|
||||||
"encoders.orly/tag"
|
"encoders.orly/tag"
|
||||||
"github.com/dgraph-io/badger/v4"
|
"github.com/dgraph-io/badger/v4"
|
||||||
@@ -45,6 +47,16 @@ func (d *D) SaveEvent(c context.Context, ev *event.E) (kc, vc int, err error) {
|
|||||||
err = errorf.E("event already exists: %0x", ev.ID)
|
err = errorf.E("event already exists: %0x", ev.ID)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// If the error is "id not found", we can proceed with saving the event
|
||||||
|
if err != nil && strings.Contains(err.Error(), "id not found in database") {
|
||||||
|
// Reset error since this is expected for new events
|
||||||
|
err = nil
|
||||||
|
} else if err != nil {
|
||||||
|
// For any other error, return it
|
||||||
|
log.E.F("error checking if event exists: %s", err)
|
||||||
|
return
|
||||||
|
}
|
||||||
// check for replacement
|
// check for replacement
|
||||||
if kind.IsReplaceable(ev.Kind) {
|
if kind.IsReplaceable(ev.Kind) {
|
||||||
// find the events and delete them
|
// find the events and delete them
|
||||||
@@ -153,6 +165,6 @@ func (d *D) SaveEvent(c context.Context, ev *event.E) (kc, vc int, err error) {
|
|||||||
return
|
return
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
log.T.F("total data written: %d bytes keys %d bytes values", kc, vc)
|
log.T.F("total data written: %d bytes keys %d bytes values for event ID %s", kc, vc, hex.Enc(ev.ID))
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user