feat(store): in-memory state machine, indexes, MemUsage
This commit is contained in:
Hermes
parent
56563465a7
commit
58a4e1db8f
1 changed files with 296 additions and 0 deletions
296
store.go
Normal file
296
store.go
Normal file
|
|
@ -0,0 +1,296 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"sort"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Store — the in-memory state machine. Hot, container-style, single-writer per pod.
|
||||
// Persistence is handled by Snapshot/Restore (see snapshot.go) and the cold-flush
|
||||
// pipeline (see flush.go). The store does NOT touch the filesystem on the hot path.
|
||||
type Store struct {
|
||||
mu sync.RWMutex // base lock; per-author/per-post locks layered on top (see locks.go)
|
||||
|
||||
posts map[string]*Post
|
||||
comments map[string]*Comment
|
||||
authors map[string]*Author
|
||||
channels map[string]*Channel
|
||||
postChannels map[string][]string // post_id -> []channel (sorted)
|
||||
mentions map[string][]*Mention
|
||||
readState map[[2]string]time.Time
|
||||
|
||||
// Indexes — derived from the maps above, maintained on every write.
|
||||
postsByBumpedAt *SortedIndex[*Post] // DESC by bumped_at
|
||||
commentsByPostPath map[string]*SortedIndex[*Comment] // post_id -> sorted by path
|
||||
|
||||
// Adaptive lock controller (see locks.go). Nil means single-writer mode.
|
||||
lockCtrl *LockController
|
||||
|
||||
// Cold-flush plumbing (see flush.go).
|
||||
flushCh chan struct{}
|
||||
flushInterval time.Duration
|
||||
flushPath string
|
||||
|
||||
// Injected for testability. nil defaults are wired in NewStore.
|
||||
now func() time.Time
|
||||
newID func() string
|
||||
}
|
||||
|
||||
// NewStore — production constructor. Wires defaults, starts the flush goroutine.
|
||||
func NewStore() *Store {
|
||||
return NewStoreWithOptions(DefaultStoreOptions())
|
||||
}
|
||||
|
||||
// StoreOptions — knobs for NewStoreWithOptions. Production uses DefaultStoreOptions.
|
||||
// Tests inject custom clock, ID generator, and a no-op flush goroutine.
|
||||
type StoreOptions struct {
|
||||
Now func() time.Time
|
||||
NewID func() string
|
||||
FlushInterval time.Duration
|
||||
FlushPath string
|
||||
StartFlush bool // false in tests, true in main
|
||||
}
|
||||
|
||||
func DefaultStoreOptions() StoreOptions {
|
||||
return StoreOptions{
|
||||
Now: func() time.Time { return time.Now().UTC() },
|
||||
NewID: defaultNewID,
|
||||
FlushInterval: 5 * time.Minute,
|
||||
FlushPath: "/mnt/nas/posts-cold",
|
||||
StartFlush: true,
|
||||
}
|
||||
}
|
||||
|
||||
func NewStoreWithOptions(opts StoreOptions) *Store {
|
||||
s := &Store{
|
||||
posts: make(map[string]*Post, 1024),
|
||||
comments: make(map[string]*Comment, 4096),
|
||||
authors: make(map[string]*Author),
|
||||
channels: make(map[string]*Channel),
|
||||
postChannels: make(map[string][]string),
|
||||
mentions: make(map[string][]*Mention),
|
||||
readState: make(map[[2]string]time.Time),
|
||||
postsByBumpedAt: NewSortedIndex[*Post](func(a, b *Post) bool { return a.BumpedAt.After(b.BumpedAt) }),
|
||||
commentsByPostPath: make(map[string]*SortedIndex[*Comment]),
|
||||
flushCh: make(chan struct{}, 1),
|
||||
flushInterval: opts.FlushInterval,
|
||||
flushPath: opts.FlushPath,
|
||||
now: opts.Now,
|
||||
newID: opts.NewID,
|
||||
lockCtrl: NewLockController(),
|
||||
}
|
||||
if opts.StartFlush {
|
||||
go s.flushLoop()
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// Now — central time helper. Injected for deterministic tests.
|
||||
func (s *Store) Now() time.Time { return s.now() }
|
||||
|
||||
// NewID — central ID generator. Injected for deterministic tests.
|
||||
func (s *Store) NewID() string { return s.newID() }
|
||||
|
||||
// MemUsage — reports approximate live bytes per type. Feeds the VOPR harness (#14)
|
||||
// and the deferred memory-budget decision (posts#1, Q7).
|
||||
type MemUsage struct {
|
||||
Posts int
|
||||
Comments int
|
||||
Authors int
|
||||
Channels int
|
||||
Mentions int
|
||||
ReadState int
|
||||
TotalApprox int
|
||||
}
|
||||
|
||||
func (s *Store) MemUsage() MemUsage {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
return MemUsage{
|
||||
Posts: len(s.posts) * approxPostSize,
|
||||
Comments: len(s.comments) * approxCommentSize,
|
||||
Authors: len(s.authors) * approxAuthorSize,
|
||||
Channels: len(s.channels) * approxChannelSize,
|
||||
Mentions: sumMentionSize(s.mentions),
|
||||
ReadState: len(s.readState) * 40, // author_id+post_id+time.Time ~40 bytes
|
||||
TotalApprox: 0,
|
||||
}
|
||||
}
|
||||
|
||||
// Approximate per-object sizes; the VOPR harness will measure precisely.
|
||||
// These are upper bounds, not accounting for interning or sync.Pool reuse.
|
||||
const (
|
||||
approxPostSize = 512
|
||||
approxCommentSize = 384
|
||||
approxAuthorSize = 256
|
||||
approxChannelSize = 192
|
||||
approxMentionSize = 256
|
||||
)
|
||||
|
||||
func sumMentionSize(m map[string][]*Mention) int {
|
||||
total := 0
|
||||
for _, v := range m {
|
||||
total += len(v) * approxMentionSize
|
||||
}
|
||||
return total
|
||||
}
|
||||
|
||||
// --- Read methods (RLock) ---
|
||||
|
||||
func (s *Store) GetPost(id string) (*Post, bool) {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
p, ok := s.posts[id]
|
||||
return p, ok
|
||||
}
|
||||
|
||||
func (s *Store) GetAuthor(id string) (*Author, bool) {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
a, ok := s.authors[id]
|
||||
return a, ok
|
||||
}
|
||||
|
||||
func (s *Store) GetChannel(slug string) (*Channel, bool) {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
c, ok := s.channels[slug]
|
||||
return c, ok
|
||||
}
|
||||
|
||||
func (s *Store) ListChannels() []*Channel {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
out := make([]*Channel, 0, len(s.channels))
|
||||
for _, c := range s.channels {
|
||||
out = append(out, c)
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].Slug < out[j].Slug })
|
||||
return out
|
||||
}
|
||||
|
||||
// ListPosts — newest-bumped first, optional channel filter. Hot path.
|
||||
func (s *Store) ListPosts(channel string, limit int) []*Post {
|
||||
if limit <= 0 || limit > 100 {
|
||||
limit = 20
|
||||
}
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
all := s.postsByBumpedAt.Items()
|
||||
out := make([]*Post, 0, limit)
|
||||
for _, p := range all {
|
||||
if p.IsDeleted {
|
||||
continue
|
||||
}
|
||||
if channel != "" {
|
||||
chs := s.postChannels[p.ID]
|
||||
if !contains(chs, channel) {
|
||||
continue
|
||||
}
|
||||
}
|
||||
out = append(out, p)
|
||||
if len(out) >= limit {
|
||||
break
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (s *Store) ListComments(postID string) []*Comment {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
idx, ok := s.commentsByPostPath[postID]
|
||||
if !ok {
|
||||
return nil
|
||||
}
|
||||
return idx.Items()
|
||||
}
|
||||
|
||||
// ListMentions — for an author, all unseen (or seen) mentions.
|
||||
func (s *Store) ListMentions(authorID string, seen bool) []*Mention {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
all := s.mentions[authorID]
|
||||
out := make([]*Mention, 0, len(all))
|
||||
for _, m := range all {
|
||||
if m.Seen == seen {
|
||||
out = append(out, m)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// --- Write methods (Lock) — direct, used by tests and bootstrap. Production
|
||||
// writes go through the Mutation interface (see mutation.go). ---
|
||||
|
||||
func (s *Store) UpsertAuthor(a *Author) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
a.UpdatedAt = s.Now()
|
||||
if _, exists := s.authors[a.ID]; !exists {
|
||||
a.CreatedAt = a.UpdatedAt
|
||||
}
|
||||
s.authors[a.ID] = a
|
||||
}
|
||||
|
||||
func (s *Store) UpsertChannel(c *Channel) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if _, exists := s.channels[c.Slug]; !exists {
|
||||
c.CreatedAt = s.Now()
|
||||
}
|
||||
s.channels[c.Slug] = c
|
||||
}
|
||||
|
||||
func (s *Store) ChannelsForPost(postID string) []string {
|
||||
s.mu.RLock()
|
||||
defer s.mu.RUnlock()
|
||||
chs := s.postChannels[postID]
|
||||
out := make([]string, len(chs))
|
||||
copy(out, chs)
|
||||
return out
|
||||
}
|
||||
|
||||
// --- Index maintenance (called under write lock by mutation Apply methods) ---
|
||||
|
||||
func (s *Store) indexInsertPost(p *Post) {
|
||||
s.postsByBumpedAt.Insert(p)
|
||||
}
|
||||
|
||||
func (s *Store) indexUpdatePost(p *Post) {
|
||||
s.postsByBumpedAt.Update(p)
|
||||
}
|
||||
|
||||
func (s *Store) indexDeletePost(p *Post) {
|
||||
s.postsByBumpedAt.Delete(p)
|
||||
}
|
||||
|
||||
func (s *Store) indexInsertComment(c *Comment) {
|
||||
idx, ok := s.commentsByPostPath[c.PostID]
|
||||
if !ok {
|
||||
idx = NewSortedIndex[*Comment](func(a, b *Comment) bool { return a.Path < b.Path })
|
||||
s.commentsByPostPath[c.PostID] = idx
|
||||
}
|
||||
idx.Insert(c)
|
||||
}
|
||||
|
||||
func (s *Store) commentsForPostIndex(postID string) *SortedIndex[*Comment] {
|
||||
return s.commentsByPostPath[postID]
|
||||
}
|
||||
|
||||
// --- helpers ---
|
||||
|
||||
func contains(xs []string, x string) bool {
|
||||
for _, v := range xs {
|
||||
if v == x {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// defaultNewID — production ID generator. 8 bytes crypto/rand -> 16 hex chars.
|
||||
func defaultNewID() string {
|
||||
return NewID()
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue