amadan

oficina / memoria Public

🤖 test: verify refresh cancellation and synchronize log capture

Independent refresh mutations proved red and restored; combined repository and race gates pass. Model review remains pending.
Co-Authored-By: Codex <noreply@openai.com>
Paul Campbell pushed by paul@keymail.dev ad41d91c3fa7b47354f51cb9d1b7b6d70b44fbe2 parent b22c65d
2 files changed, +119 −7
  • internal/memoria/stale_refresh_test.go +97 −5
  • internal/memoriatest/connector_test.go +22 −2

CI passed — run details

diff --git a/internal/memoria/stale_refresh_test.go b/internal/memoria/stale_refresh_test.go
index d8d3ac8..1260916 100644
--- a/internal/memoria/stale_refresh_test.go
+++ b/internal/memoria/stale_refresh_test.go
@@ -17,6 +17,7 @@ import (
)
type staleFixture struct {
+ released atomic.Bool
failFeed atomic.Bool
fixedRuling atomic.Value
store *Store
@@ -101,7 +102,7 @@ func newStaleFixture(t *testing.T) *staleFixture {
select {
case <-f.release:
default:
- close(f.release)
+ f.unblock()
}
f.wait(t)
srv.Close()
@@ -117,6 +118,12 @@ func newStaleFixture(t *testing.T) *staleFixture {
return f
}
+func (f *staleFixture) unblock() {
+ if f.released.CompareAndSwap(false, true) {
+ close(f.release)
+ }
+}
+
func (f *staleFixture) search(ctx context.Context) *httptest.ResponseRecorder {
rec := httptest.NewRecorder()
req := httptest.NewRequest(http.MethodGet, "/search?q=quarterly", nil).WithContext(ctx)
@@ -124,8 +131,14 @@ func (f *staleFixture) search(ctx context.Context) *httptest.ResponseRecorder {
return rec
}
+// MUTATION-TESTED 2026-09-07, independently selected by the coordinator:
+// removing the hook, inheriting cancellation, refreshing every ruling,
+// bypassing the join, retaining completed jobs and waiting on the reader
+// each failed on the named behavior, then was reverted separately.
func TestStaleSearchStartsBoundedCatchupWithoutWaiting(t *testing.T) {
f := newStaleFixture(t)
+ watchdog := time.AfterFunc(2*sibling.Deadline, f.unblock)
+ defer watchdog.Stop()
started := time.Now()
result := f.search(context.Background())
if time.Since(started) > sibling.Deadline {
@@ -139,7 +152,7 @@ func TestStaleSearchStartsBoundedCatchupWithoutWaiting(t *testing.T) {
case <-time.After(time.Second):
t.Fatal("STALE said refetching but no S1 catchup started")
}
- close(f.release)
+ f.unblock()
limit := time.Now().Add(2 * time.Second)
for {
cursor, err := f.store.cursor(context.Background(), "Docs")
@@ -211,7 +224,7 @@ func TestStaleRefreshCoalescesWithWakeAndRepeatedRequests(t *testing.T) {
if f.calls.Load() != 1 {
t.Errorf("S1 calls while blocked%d", f.calls.Load())
}
- close(f.release)
+ f.unblock()
select {
case <-warmDone:
case <-time.After(3 * time.Second):
@@ -232,7 +245,7 @@ func TestStaleRefreshOutlivesCancelledReader(t *testing.T) {
t.Fatal("refresh not started")
}
cancel()
- close(f.release)
+ f.unblock()
f.wait(t)
cursor, err := f.store.cursor(context.Background(), "Docs")
if err != nil || cursor != "c73" {
@@ -249,7 +262,7 @@ func TestStaleRefreshFailurePreservesCursorAndCanRetry(t *testing.T) {
case <-time.After(time.Second):
t.Fatal("refresh not started")
}
- close(f.release)
+ f.unblock()
f.wait(t)
cursor, err := f.store.cursor(context.Background(), "Docs")
if err != nil || cursor != "c19" {
@@ -288,3 +301,82 @@ func TestOnlyStaleTriggersRefresh(t *testing.T) {
})
}
}
+
+func TestStaleRefreshJoiningCancelledWarmCanRetry(t *testing.T) {
+ f := newStaleFixture(t)
+ owner := f.store.custody.(*Siblings)
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ warmDone := make(chan struct{})
+ go func() { owner.Warm(ctx, f.store, f.app.logger); close(warmDone) }()
+ select {
+ case <-f.entered:
+ case <-time.After(time.Second):
+ t.Fatal("warm not started")
+ }
+ result := f.search(context.Background())
+ if strings.Contains(result.Body.String(), "OldQuarterlyCanary") {
+ t.Fatal("joining warm exposed stale text")
+ }
+ cancel()
+ select {
+ case <-warmDone:
+ case <-time.After(time.Second):
+ t.Fatal("warm cancellation did not return")
+ }
+ f.wait(t)
+ cursor, err := f.store.cursor(context.Background(), "Docs")
+ if err != nil || cursor != "c19" {
+ t.Errorf("cancelled warm moved cursor: %q %v", cursor, err)
+ }
+ f.search(context.Background())
+ select {
+ case <-f.entered:
+ case <-time.After(time.Second):
+ t.Fatal("next stale request did not retry after cancelled warm")
+ }
+ f.unblock()
+ f.wait(t)
+ cursor, err = f.store.cursor(context.Background(), "Docs")
+ if err != nil || cursor != "c73" {
+ t.Errorf("retry did not catch up: %q %v", cursor, err)
+ }
+}
+
+func TestDocumentReadersAlsoTriggerStaleRefresh(t *testing.T) {
+ for _, kind := range []string{"list", "document"} {
+ t.Run(kind, func(t *testing.T) {
+ f := newStaleFixture(t)
+ docs, err := f.store.Summaries(context.Background())
+ if err != nil || len(docs) != 1 {
+ t.Fatal("fixture missing")
+ }
+ subject := suite.Subject{Kind: suite.SubjectAddress, Value: "reader@example.test"}
+ var notices []string
+ switch kind {
+ case "list":
+ got, n, e := f.store.ReaderSummaries(context.Background(), subject)
+ if e != nil || len(got) != 0 {
+ t.Fatal("stale list did not withhold")
+ }
+ notices = n
+ case "document":
+ got, n, e := f.store.ReadDocument(context.Background(), subject, docs[0].ID)
+ if e == nil || len(got.Passages) != 0 {
+ t.Fatal("stale document did not withhold")
+ }
+ notices = n
+ }
+ if len(notices) == 0 {
+ t.Fatal("stale reader was silent")
+ }
+ select {
+ case <-f.entered:
+ case <-time.After(time.Second):
+ t.Fatal("reader promised refetch but started none")
+ }
+ f.unblock()
+ f.wait(t)
+ })
+ }
+}
diff --git a/internal/memoriatest/connector_test.go b/internal/memoriatest/connector_test.go
index 11d394a..4791092 100644
--- a/internal/memoriatest/connector_test.go
+++ b/internal/memoriatest/connector_test.go
@@ -8,6 +8,7 @@ import (
"net/url"
"path/filepath"
"strings"
+ stdsync "sync"
"testing"
"time"
@@ -125,11 +126,23 @@ func (f *fakeSource) rename(id, was, now string) {
}
}
+type lockedLogBuffer struct {
+ mu stdsync.Mutex
+ text strings.Builder
+}
+
+func (b *lockedLogBuffer) Write(p []byte) (int, error) {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ return b.text.Write(p)
+}
+func (b *lockedLogBuffer) String() string { b.mu.Lock(); defer b.mu.Unlock(); return b.text.String() }
+
// connectorApp builds an app with a configured connector and a fake
// Notion behind it.
-func connectorApp(t *testing.T, src *fakeSource) (http.Handler, testApp, int64, *strings.Builder) {
+func connectorApp(t *testing.T, src *fakeSource) (http.Handler, testApp, int64, *lockedLogBuffer) {
t.Helper()
- var logs strings.Builder
+ var logs lockedLogBuffer
logger := slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelDebug}))
d, err := db.Open(filepath.Join(t.TempDir(), "app.db"), logger)
if err != nil {
@@ -457,6 +470,13 @@ func TestNoNotionContentInLogs(t *testing.T) {
sync(t, h, app, cid)
get(t, h, "/connectors/notion")
+ deadline := time.Now().Add(5 * time.Second)
+ for !strings.Contains(logs.String(), "sync finished") {
+ if time.Now().After(deadline) {
+ t.Fatal("final sync log was not observed")
+ }
+ time.Sleep(time.Millisecond)
+ }
got := logs.String()
if strings.Contains(got, secret) {
t.Errorf("a log line carries Notion content:\n%s", got)