package spindle
import (
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"runtime/debug"
"sync"
"time"
"tangled.org/core/api/org_tangled"
"tangled.org/core/jetstream"
)
const (
ingestShardCount = 8
ingestShardQueue = 256
)
var errIngestStopped = errors.New("the ingest shards stopped")
type ingestTicket interface {
Done()
Drop()
}
type ingestItem struct {
ctx context.Context
e *jetstream.Event
ticket ingestTicket
}
// same repo always lands on the same shard, that's where ordering comes from
type ingestShards struct {
apply func(context.Context, *jetstream.Event) error
queues [ingestShardCount]chan ingestItem
wg sync.WaitGroup
// stop takes the write lock, which keeps any send from landing after it
mu sync.RWMutex
stopped bool
quit chan struct{}
quitOne sync.Once
}
func newIngestShards(ctx context.Context, apply func(context.Context, *jetstream.Event) error) *ingestShards {
sh := &ingestShards{apply: apply, quit: make(chan struct{})}
for i := range sh.queues {
sh.queues[i] = make(chan ingestItem, ingestShardQueue)
sh.wg.Go(func() { sh.run(ctx, sh.queues[i]) })
}
return sh
}
// declarations shard by the repo they name, which puts them behind its other events
func ingestShardKey(e *jetstream.Event) string {
if e.Commit != nil && e.Commit.Collection == org_tangled.RepoDeclarationNSID && e.Commit.Operation != jetstream.CommitOperationDelete {
var declaration org_tangled.RepoDeclaration
if json.Unmarshal(e.Commit.Record, &declaration) == nil && declaration.Repo != "" {
return declaration.Repo
}
}
return e.Did
}
func ingestShardOf(key string) int {
return int(fnv32a(key) % ingestShardCount)
}
func (sh *ingestShards) dispatch(ctx context.Context, e *jetstream.Event, ticket ingestTicket) error {
sh.mu.RLock()
defer sh.mu.RUnlock()
if sh.stopped {
ticket.Drop()
return errIngestStopped
}
select {
case sh.queues[ingestShardOf(ingestShardKey(e))] <- ingestItem{ctx: ctx, e: e, ticket: ticket}:
return nil
case <-ctx.Done():
ticket.Drop()
return ctx.Err()
case <-sh.quit:
ticket.Drop()
return errIngestStopped
}
}
func (sh *ingestShards) run(ctx context.Context, queue chan ingestItem) {
for {
select {
case it := <-queue:
// even after cancel, select can pick a queued event, check here
if ctx.Err() != nil {
it.ticket.Drop()
continue
}
sh.handle(it)
case <-ctx.Done():
sh.quitOne.Do(func() { close(sh.quit) })
sh.mu.Lock()
sh.stopped = true
sh.mu.Unlock()
for {
select {
case it := <-queue:
it.ticket.Drop()
default:
return
}
}
}
}
}
// without this a panic kills the process before the cursor moves, and every start dies on it again
func recovered(what string, fn func() error) (err error) {
defer func() {
if p := recover(); p != nil {
slog.Error("handler panicked, skipping it", "what", what, "panic", p, "stack", string(debug.Stack()))
err = fmt.Errorf("panic in %s: %v: %w", what, p, errHandlerPanicked)
}
}()
return fn()
}
var errHandlerPanicked = errors.New("the handler panicked")
const ingestAttempts = 5
// failures are mostly a busy db, which makes a few retries worth it
var ingestRetryDelay = 100 * time.Millisecond
func (sh *ingestShards) handle(it ingestItem) {
var err error
for attempt := range ingestAttempts {
err = recovered("ingest shard", func() error { return sh.apply(it.ctx, it.e) })
if err == nil || errors.Is(err, errHandlerPanicked) || it.ctx.Err() != nil {
break
}
if attempt < ingestAttempts-1 {
select {
case <-time.After(ingestRetryDelay << attempt):
case <-it.ctx.Done():
}
}
}
// the cursor stays behind a failed event, which means the next start reads it again
if err != nil && !errors.Is(err, errHandlerPanicked) && it.ctx.Err() == nil {
slog.Error("giving up on an event, the cursor stays behind it", "did", it.e.Did, "err", err)
it.ticket.Drop()
return
}
finishTicket(it.ctx, it.ticket, err)
}
// on cancel, drop rather than done, or the cursor would move past it
func finishTicket(ctx context.Context, t ingestTicket, err error) {
if err != nil && ctx.Err() != nil {
t.Drop()
return
}
t.Done()
}