kami/Maven
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

sweep: dedupe memory_vectors scan, fix two swallowed errors (V-581)

Browse Source 
allMemVectorMetas (memory.go) replaces the identical query-then-scan
block ReembedAll and RepairFactVectors each had for reading id+meta
out of memory_vectors — same query, same json.Unmarshal, different
structs built from the result.

Two RowsAffected() errors were silently dropped with `_`, inconsistent
with every other call site in the same files: AcceptProposedRoutine
now wraps the error instead of treating it as zero rows, and MarkAcked
had it stranded behind a dead branch (both arms returned nil) removed
along with the swallowed error.

No behavior change; internal/store and internal/memory pass with
-race.
This commit is contained in:
claude
2026-08-06 03:13:47 +04:00
parent 1fa14e95a4
commit 5bca435146
5 changed files with 62 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/store/ack.go
+4 -6
View File
@@ -55,17 +55,15 @@ func (s *Store) LastSent(ctx context.Context, key string) (time.Time, error) {
// alarm (voice acknowledgment, Telegram callback, etc.).
func (s *Store) MarkAcked(ctx context.Context, key string) error {
now := time.Now()
res, err := s.db.ExecContext(ctx,
// No pending nudges is not an error — already acked or never sent — so the
// rows-affected count is not read at all: every outcome below this line is
// the same nil.
_, err := s.db.ExecContext(ctx,
`UPDATE nudges SET outcome = 'acted', outcome_ts = ?
WHERE rule = ? AND channel = 'telegram' AND outcome = 'pending'`,
now.UnixMilli(), key)
if err != nil {
return fmt.Errorf("mark acked %s: %w", key, err)
}
n, _ := res.RowsAffected()
if n == 0 {
// no pending nudges — already acked or never sent; not an error.
return nil
}
return nil
}
internal/store/backfill.go
+5 -20
View File
@@ -2,7 +2,6 @@ package store
import (
"context"
"encoding/json"
"fmt"
"time"
)
@@ -103,30 +102,16 @@ func (s *Store) ReembedAll(ctx context.Context, currentID string, embed EmbedFun
id, text, kind string
}
var vecs []vecRow
rows, err = tx.QueryContext(ctx, `SELECT id, meta FROM memory_vectors`)
memRows, err := allMemVectorMetas(ctx, tx)
if err != nil {
return res, fmt.Errorf("reembed: read memory vectors: %w", err)
return res, fmt.Errorf("reembed: %w", err)
}
for rows.Next() {
var id, metaJSON string
if err := rows.Scan(&id, &metaJSON); err != nil {
rows.Close()
return res, fmt.Errorf("reembed: memory row: %w", err)
}
meta := map[string]string{}
if err := json.Unmarshal([]byte(metaJSON), &meta); err != nil {
rows.Close()
return res, fmt.Errorf("reembed: meta for %q: %w", id, err)
}
if meta["text"] == "" {
for _, v := range memRows {
if v.Meta["text"] == "" {
res.NoText++
continue
}
vecs = append(vecs, vecRow{id: id, text: meta["text"], kind: meta["type"]})
}
rows.Close()
if err := rows.Err(); err != nil {
return res, fmt.Errorf("reembed: memory vectors: %w", err)
vecs = append(vecs, vecRow{id: v.ID, text: v.Meta["text"], kind: v.Meta["type"]})
}
for _, v := range vecs {
internal/store/factvectors.go
+6 -20
View File
@@ -64,9 +64,9 @@ func (s *Store) RepairFactVectors(ctx context.Context, embed EmbedFunc) (FactVec
return res, nil
}
rows, err := s.db.QueryContext(ctx, `SELECT id, meta FROM memory_vectors`)
memRows, err := allMemVectorMetas(ctx, s.db)
if err != nil {
return res, fmt.Errorf("repair fact vectors: read: %w", err)
return res, fmt.Errorf("repair fact vectors: %w", err)
}
type factVec struct {
id, key string
@@ -75,33 +75,19 @@ func (s *Store) RepairFactVectors(ctx context.Context, embed EmbedFunc) (FactVec
}
var vecs []factVec
newest := map[string]int64{} // key → newest ts seen for it
for rows.Next() {
var id, metaJSON string
if err := rows.Scan(&id, &metaJSON); err != nil {
rows.Close()
return res, fmt.Errorf("repair fact vectors: row: %w", err)
}
meta := map[string]string{}
if err := json.Unmarshal([]byte(metaJSON), &meta); err != nil {
rows.Close()
return res, fmt.Errorf("repair fact vectors: meta for %q: %w", id, err)
}
if meta["type"] != "fact" {
for _, v := range memRows {
if v.Meta["type"] != "fact" {
continue
}
key, ts, ok := splitFactVectorID(id)
key, ts, ok := splitFactVectorID(v.ID)
if !ok {
continue
}
vecs = append(vecs, factVec{id: id, key: key, meta: meta, ts: ts})
vecs = append(vecs, factVec{id: v.ID, key: key, meta: v.Meta, ts: ts})
if ts > newest[key] {
newest[key] = ts
}
}
rows.Close()
if err := rows.Err(); err != nil {
return res, fmt.Errorf("repair fact vectors: rows: %w", err)
}
for _, v := range vecs {
drop := v.ts < newest[v.key]
internal/store/memory.go
+43
View File
@@ -169,6 +169,49 @@ func (m *MemoryStore) DeletePrefix(ctx context.Context, prefix string) (int64, e
return n, nil
}
// memVectorRow is one memory_vectors row with its meta blob decoded — the
// shape both ReembedAll (backfill.go) and RepairFactVectors (factvectors.go)
// read the whole table as, before each decides what to do with a row on its
// own terms (one keys off meta["text"], the other off meta["type"] and the
// id's embedded key/timestamp). Query-then-scan was duplicated across the two
// before this, id-for-id.
type memVectorRow struct {
ID string
Meta map[string]string
}
// queryContexter is the common surface *sql.DB and *sql.Tx share that
// allMemVectorMetas needs. ReembedAll reads inside a transaction so its
// migration is atomic; RepairFactVectors reads directly off the db handle.
type queryContexter interface {
QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
}
// allMemVectorMetas reads every memory_vectors row and decodes its meta blob.
func allMemVectorMetas(ctx context.Context, q queryContexter) ([]memVectorRow, error) {
rows, err := q.QueryContext(ctx, `SELECT id, meta FROM memory_vectors`)
if err != nil {
return nil, fmt.Errorf("read memory vectors: %w", err)
}
defer rows.Close()
var out []memVectorRow
for rows.Next() {
var id, metaJSON string
if err := rows.Scan(&id, &metaJSON); err != nil {
return nil, fmt.Errorf("memory vector row: %w", err)
}
meta := map[string]string{}
if err := json.Unmarshal([]byte(metaJSON), &meta); err != nil {
return nil, fmt.Errorf("meta for %q: %w", id, err)
}
out = append(out, memVectorRow{ID: id, Meta: meta})
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("memory vectors: %w", err)
}
return out, nil
}
// escapeLike neutralises the LIKE wildcards in a literal prefix.
func escapeLike(s string) string {
r := strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`)
internal/store/proposed_routines.go
+4 -1
View File
@@ -187,7 +187,10 @@ func (s *Store) AcceptProposedRoutine(ctx context.Context, id int64, ts time.Tim
if err != nil {
return fmt.Errorf("accept proposed routine: %w", err)
}
n, _ := res.RowsAffected()
n, err := res.RowsAffected()
if err != nil {
return fmt.Errorf("accept proposed routine: rows affected: %w", err)
}
if n == 0 {
return fmt.Errorf("%w: id=%d not in 'proposed' status", ErrProposedRoutineNotFound, id)
}