Skip to content

refactor(core): collapse speculative flow-store scaffolding #47

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lao merged 2 commits into from
Jul 30, 2026
Merged

refactor(core): collapse speculative flow-store scaffolding #47

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4e46df2
refactor(core): collapse speculative flow-store scaffolding
lao Jul 30, 2026
f4bf5c6
test(core): route flow_test state access through locked helpers
lao Jul 30, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
216 changes: 53 additions & 163 deletions internal/core/conversation.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,19 +2,12 @@ package core

import (
"context"
"log/slog"
"maps"
"strings"
"sync"
"time"
)

// conversationShards is the fixed number of striped locks the conversationManager
// holds. It bounds lock memory by construction: there is no per-key lock map that
// grows with distinct conversations. Two keys that hash to the same shard
// serialize unnecessarily, but with this many shards that is rare and brief.
const conversationShards = 256

// defaultSweepInterval is how often the background sweeper reaps expired flow
// state when started without an explicit interval.
const defaultSweepInterval = time.Minute
Expand All @@ -27,20 +20,13 @@ const defaultSweepInterval = time.Minute
const defaultFlowTimeout = 10 * time.Minute

// ConversationState is the per-conversation flow state the engine carries between
// messages. It is flat and serializable by design so a future durable Store can
// persist it without migration; secret answers are excluded from any serialized
// form (see flow.go). Version bumps on every Set within a key's lifetime; it is
// unused by the single-instance v1 (the in-process striped locks suffice) and is
// exercised from day one so the eventual Store-backed compare-and-swap path needs
// no state change. It is not preserved across a delete-and-recreate of the same
// key (a swept conversation that restarts begins again at 1) — the multi-instance
// CAS path owns cross-lifetime versioning when it lands.
// messages, keyed per user+channel and held in memory for the life of the
// process (a restart loses every in-flight flow).
type ConversationState struct {
FlowID string
Step int
Answers map[string]string
ExpiresAt time.Time
Version uint64
}

// isExpired reports whether s has a set expiry that is at or before now. A zero
Expand All @@ -49,158 +35,62 @@ func (s ConversationState) isExpired(now time.Time) bool {
return !s.ExpiresAt.IsZero() && !s.ExpiresAt.After(now)
}

// ConversationStore persists per-conversation flow state by key. The default
// implementation is in-memory (memConversationStore); the interface is the seam a
// durable backend (e.g. Redis) drops into later. The conversationManager
// serializes logical access per key with its striped locks, but a store is still
// touched from the background sweeper, so implementations must be safe for
// concurrent use.
type ConversationStore interface {
Get(key string) (ConversationState, bool)
Set(key string, state ConversationState)
Delete(key string)
}

// expirer is the optional sweeper capability a ConversationStore may implement: it
// reports the keys eligible for reaping at a given instant. memConversationStore
// implements it; a backend with native TTLs (e.g. Redis) need not, and the
// manager simply skips background sweeping for such a store. This mirrors the
// AttachmentResolver optional-capability idiom in core.go.
type expirer interface {
expiredKeys(now time.Time) []string
}

// memConversationStore is the default in-memory ConversationStore: a map guarded
// by its own RWMutex, held only for the map access itself and never across a
// validator. It is the volatile home for in-flight flows; a process restart loses
// everything in it.
type memConversationStore struct {
mu sync.RWMutex
m map[string]ConversationState
}

func newMemConversationStore() *memConversationStore {
return &memConversationStore{m: make(map[string]ConversationState)}
}

func (s *memConversationStore) Get(key string) (ConversationState, bool) {
s.mu.RLock()
defer s.mu.RUnlock()
st, ok := s.m[key]
return st, ok
}

// Set stores state under key, bumping Version past the previous value for the key
// (1 for a fresh key) so the field is always monotonic and correct.
func (s *memConversationStore) Set(key string, state ConversationState) {
s.mu.Lock()
defer s.mu.Unlock()
state.Version = s.m[key].Version + 1
s.m[key] = state
}

func (s *memConversationStore) Delete(key string) {
s.mu.Lock()
defer s.mu.Unlock()
delete(s.m, key)
}

// expiredKeys returns the keys whose ExpiresAt is set and at or before now. A zero
// ExpiresAt never expires.
func (s *memConversationStore) expiredKeys(now time.Time) []string {
s.mu.RLock()
defer s.mu.RUnlock()
var expired []string
for k, st := range s.m {
if st.isExpired(now) {
expired = append(expired, k)
}
}
return expired
}

// conversationManager owns the in-process coordination for flows: a fixed-size
// striped lock set plus the background TTL sweeper. The ConversationStore behind
// it owns only persistence. The user validator runs UNDER a shard lock, never
// inside a store call, so a durable store can later sit behind compare-and-swap
// without relocating the validator (which is arbitrary user Go code).
// conversationManager owns the in-process coordination for flows: the per-
// conversation state map plus the background TTL sweeper, both guarded by one
// mutex.
//
// ponytail: single global mutex, not per-key striped locks. It serializes every
// conversation's state transition — and the user validator that runs under it,
// per Validate's "keep it fast" contract — against every other conversation. For
// a single-instance in-memory chat bot that is ample; shard the lock by key only
// if flow throughput is ever a measured bottleneck.
type conversationManager struct {
store ConversationStore
locks [conversationShards]sync.Mutex
mu sync.Mutex
m map[string]ConversationState
}

func newConversationManager() *conversationManager {
return &conversationManager{store: newMemConversationStore()}
}

// shardFor returns the lock guarding key. The same key always maps to the same
// shard. FNV-1a is inlined over the key's bytes to avoid the per-call hasher heap
// allocation fnv.New32a() incurs (it returns an interface).
func (m *conversationManager) shardFor(key string) *sync.Mutex {
const (
offset32 = 2166136261
prime32 = 16777619
)
h := uint32(offset32)
for i := 0; i < len(key); i++ {
h ^= uint32(key[i])
h *= prime32
}
return &m.locks[h%conversationShards]
return &conversationManager{m: make(map[string]ConversationState)}
}

// withLock runs fn while holding key's shard lock, releasing it even if fn
// panics. The panic propagates to dispatch's recover; the shard stays usable for
// the next operation on any key it guards.
func (m *conversationManager) withLock(key string, fn func()) {
mu := m.shardFor(key)
mu.Lock()
defer mu.Unlock()
// withLock runs fn while holding the manager mutex, releasing it even if fn
// panics (the panic propagates to dispatch's recover). The get/set/del helpers
// assume this lock is held.
func (m *conversationManager) withLock(fn func()) {
m.mu.Lock()
defer m.mu.Unlock()
fn()
}

// sweepRecovered runs sweep with a recover so a panic in a custom
// ConversationStore cannot take down the sweeper goroutine (and with it the
// process), mirroring the recover that guards dispatch in core.go. logger (may be
// nil → slog.Default) is passed in rather than held on the manager so a sweeper
// goroutine never shares mutable diagnostics state with a concurrent reconnect.
func (m *conversationManager) sweepRecovered(now time.Time, logger *slog.Logger) {
defer func() {
if r := recover(); r != nil {
if logger == nil {
logger = slog.Default()
}
logger.Error("botbooter: recovered from panic in conversation sweeper", "panic", r)
}
}()
m.sweep(now)
// get, set and del are plain map operations; the caller MUST hold m.mu (via
// withLock).
func (m *conversationManager) get(key string) (ConversationState, bool) {
st, ok := m.m[key]
return st, ok
}

// sweep deletes every expired entry as of now. It takes each key's shard lock and
// re-checks expiry under it, so it can never race a concurrent advance that just
// refreshed the entry's TTL. A store without the expirer capability is a no-op.
func (m *conversationManager) set(key string, st ConversationState) { m.m[key] = st }

func (m *conversationManager) del(key string) { delete(m.m, key) }

// sweep deletes every entry expired as of now. It holds the manager mutex across
// the whole scan, so it can never race a concurrent advance that just refreshed
// an entry's TTL (deleting during a map range is safe in Go).
func (m *conversationManager) sweep(now time.Time) {
ex, ok := m.store.(expirer)
if !ok {
return
}
for _, key := range ex.expiredKeys(now) {
m.withLock(key, func() {
if st, ok := m.store.Get(key); ok && st.isExpired(now) {
m.store.Delete(key)
m.withLock(func() {
for key, st := range m.m {
if st.isExpired(now) {
delete(m.m, key)
}
})
}
}
})
}

// startSweeper runs sweep on an interval until ctx is done, then closes the
// returned channel so a caller can observe that the goroutine has exited — making
// the sweeper leak-free and testable. A non-positive interval falls back to
// defaultSweepInterval. logger (may be nil) routes sweeper panic recovery through
// the Bot's configured diagnostics sink; it is captured by the goroutine, not
// stored on the manager, so a reconnect that starts a fresh sweeper never races a
// lingering old one over shared state.
func (m *conversationManager) startSweeper(ctx context.Context, interval time.Duration, logger *slog.Logger) <-chan struct{} {
// defaultSweepInterval.
func (m *conversationManager) startSweeper(ctx context.Context, interval time.Duration) <-chan struct{} {
if interval <= 0 {
interval = defaultSweepInterval
}
Expand All @@ -214,7 +104,7 @@ func (m *conversationManager) startSweeper(ctx context.Context, interval time.Du
case <-ctx.Done():
return
case <-ticker.C:
m.sweepRecovered(time.Now(), logger)
m.sweep(time.Now())
}
}
}()
Expand Down Expand Up @@ -294,19 +184,19 @@ func (b *Bot) sendFlowMessage(ctx context.Context, channelID, text string) {
}
}

// start begins flow for msg's conversation. Under the shard lock it records the
// start begins flow for msg's conversation. Under the lock it records the
// initial state if and only if none exists (set-if-absent); after releasing the
// lock it sends the first prompt. A losing racer — state already present — is a
// no-op, and its trigger message is dropped, never consumed as the first answer.
func (m *conversationManager) start(ctx context.Context, b *Bot, msg *Message, flow *Flow) {
key := conversationKey(msg)

var started bool
m.withLock(key, func() {
if _, ok := m.store.Get(key); ok {
m.withLock(func() {
if _, ok := m.get(key); ok {
return // already active for this conversation; drop the trigger
}
m.store.Set(key, ConversationState{
m.set(key, ConversationState{
FlowID: flow.id,
Step: 0,
Answers: map[string]string{},
Expand All @@ -333,7 +223,7 @@ func (m *conversationManager) advance(ctx context.Context, b *Bot, msg *Message)
handled bool
action func()
)
m.withLock(key, func() {
m.withLock(func() {
handled, action = m.transitionLocked(ctx, b, key, msg, now)
})

Expand All @@ -353,7 +243,7 @@ func (m *conversationManager) advance(ctx context.Context, b *Bot, msg *Message)
// reaped but reported handled (consuming the trigger) and runs the optional
// OnTimeout. The asymmetry is deliberate — see the design spec.
func (m *conversationManager) transitionLocked(ctx context.Context, b *Bot, key string, msg *Message, now time.Time) (handled bool, action func()) {
state, ok := m.store.Get(key)
state, ok := m.get(key)
if !ok {
return false, nil // no active flow; dispatch continues to the command table
}
Expand All @@ -362,7 +252,7 @@ func (m *conversationManager) transitionLocked(ctx context.Context, b *Bot, key
if !ok {
// State outlived its flow registration (e.g. a renamed flow); reap and
// fall through rather than panic.
m.store.Delete(key)
m.del(key)
return false, nil
}

Expand All @@ -371,12 +261,12 @@ func (m *conversationManager) transitionLocked(ctx context.Context, b *Bot, key
// eaten by dispatch's recover but leave the state in place to wedge every
// later message until its TTL.)
if state.Step < 0 || state.Step >= len(flow.steps) {
m.store.Delete(key)
m.del(key)
return false, nil
}

if state.isExpired(now) {
m.store.Delete(key)
m.del(key)
if flow.onTimeout != nil {
return true, func() { flow.onTimeout(ctx, b, msg) }
}
Expand All @@ -391,7 +281,7 @@ func (m *conversationManager) transitionLocked(ctx context.Context, b *Bot, key

// The cancel word shadows every step, so it precedes validation.
if flow.cancelWord != "" && content == flow.cancelWord {
m.store.Delete(key)
m.del(key)
if flow.onCancel != nil {
return true, func() { flow.onCancel(ctx, b, msg) }
}
Expand All @@ -406,7 +296,7 @@ func (m *conversationManager) transitionLocked(ctx context.Context, b *Bot, key
// quiet lets it expire.
slideTTL := func() {
state.ExpiresAt = now.Add(flow.timeoutOrDefault())
m.store.Set(key, state)
m.set(key, state)
}

// Empty/whitespace answers are non-answers: re-prompt without storing an answer.
Expand Down Expand Up @@ -451,7 +341,7 @@ func (m *conversationManager) transitionLocked(ctx context.Context, b *Bot, key
// engine reusing the underlying map.
if state.Step == len(flow.steps)-1 {
completed := Answers(maps.Clone(answers))
m.store.Delete(key)
m.del(key)
return true, func() { flow.onComplete(ctx, b, msg, completed) }
}

Expand Down
Loading
Loading