amadan

oficina / memoria Public

Separate bounded sibling ingest from search deadlines

Co-Authored-By: Codex <codex@openai.com>
Paul Campbell pushed by paul@keymail.dev 80f813b2d2301570fc612a6a58e26b9e5fbb20fc parent a012c8f
6 files changed, +314 −2
  • docs/superpowers/plans/2026-09-07-sibling-ingest-budget.md +69 −0
  • internal/memoria/custody.go +2 −0
  • internal/memoria/ingest_budget_test.go +77 −0
  • internal/memoria/sibling/feed.go +7 −2
  • internal/memoria/sibling/ingest_deadline_test.go +154 −0
  • internal/memoria/siblingsync.go +5 −0
diff --git a/docs/superpowers/plans/2026-09-07-sibling-ingest-budget.md b/docs/superpowers/plans/2026-09-07-sibling-ingest-budget.md
new file mode 100644
index 0000000..068ab76
--- /dev/null
+++ b/docs/superpowers/plans/2026-09-07-sibling-ingest-budget.md
@@ -0,0 +1,69 @@
+# Separate ingest and search deadlines
+
+Status: DRAFT implementation, unreviewed. No merge or deployment.
+LLM implementation plan, 2026-09-07. Baseline 6a1e05b (approved dependency pin),
+branch fix/s1-deadline, private Memoria discussion 5. Custody regression branch
+e4b178f remains separate.
+
+## Settled behavior and evidence
+
+Memoria §6.5 sets 800ms per sibling for the S2 search-page authorization round.
+It expects asker-less S1 wake polling to spend cold-start latency outside a
+reader's request. Feed and Content both currently apply the S2 Deadline constant.
+A real HTTP fixture returning valid S1 JSON or S3 content after 1100ms fails at
+800ms with a caller deadline of 3s. TestIngestCanFinishAfterTheSearchDeadline
+records both failures.
+
+TestIngestRetainsCallerCancellation passes for S1/S3 both before headers and
+while reading the response body. The server observes cancellation; returned
+errors retain context.Canceled without URL, item ID or cursor. Evidence is
+/tmp/oficina-memoria-s1-review/baseline.log.
+
+No explicit ingest timeout number exists in the binding spec. The 5s budget in
+§7.3 concerns loading the sealed index, not sibling ingestion. The suite HTTP
+client at 1ac4323 has no Client.Timeout. WarmSiblings uses context.WithoutCancel
+because polling outlives its initiating request. Removing the wrappers without
+a replacement can hold a wake forever: maxFeedPages bounds pages, not time.
+
+## Proposed implementation, pending design review
+
+Keep S2 Deadline=800ms and its concurrency, UNKNOWN and cancellation behavior.
+Introduce a separate 45s ceiling for each S1/S3 request, covering connection,
+headers and the bounded response body. Keep the caller context, so an earlier
+deadline or explicit cancellation wins. Current platform source df7d1d25 gives wakeWaitBound=30s,
+spawnHealthTimeout=30s, backhaulDialTimeout=5s and backhaulHeaderTimeout=30s.
+45s allows that wake/header bound plus connection and body-transfer margin;
+it is an engineering ceiling, not a measured cold-wake capacity claim. Measure actual cold sibling latency at canary before shipping;
+if insufficient, change the reviewed cap with evidence.
+
+Bound the whole concurrent Warm call and each standalone Store.Poll to 120s, inherited by every network and page
+operation. Per-request ceilings alone allow 50 pages times many content calls
+to retain a wake excessively. Keep maxFeedPages as the independent loop cap.
+The wake stays independent of the first browser request but gains a finite poll
+lifetime. No periodic wake timer is added.
+
+On timeout use the existing error path, never an empty feed/body or a successful
+PolledAt. Keep cursor advancement after a fully applied page. Completed earlier
+pages may already have advanced; partial pages retry idempotently next poll.
+Do not change S3 404 handling, deletion rules, seals, notices or authorization.
+Post-commit derived-index refresh retains its existing WithoutCancel consistency
+rule; its obligation differs from a network timeout. No payload/URL logging.
+
+## Required verification
+
+- The real HTTP 1100ms feed/content regressions become green.
+- Caller cancellation and shorter deadlines stop header/body waits.
+- Successive slow content calls cannot reset the whole-poll budget; an incomplete
+ page leaves its cursor and PolledAt unchanged.
+- A stalled sibling cannot block another sibling's independent wake poll.
+- Existing S2 literal 800ms and concurrent-page regressions pass unchanged.
+- Inspect effective transport deadlines to prove finite production ceilings
+ without waiting 45/120s in tests; pair with real HTTP timing/cancellation cases.
+- Mutation-prove recoupling Feed/Content to S2 Deadline, discarding cancellation,
+ and resetting the total budget per row. Record observed mutations and dates
+ in test comments when implementation lands. Run mutations only in this isolated worktree, restore before committing.
+
+Root authorized this bounded technical choice as draft implementation; Fable
+review of design and whole branch is required before approval or integration. No review call
+before the 01:50 Europe/Dublin reset and root allocation. No merge/deployment;
+the full release still needs custody fixes, rollback controls and other findings.
diff --git a/internal/memoria/custody.go b/internal/memoria/custody.go
index bebba72..b259c6b 100644
--- a/internal/memoria/custody.go
+++ b/internal/memoria/custody.go
@@ -108,6 +108,8 @@ func (s *Siblings) Rule(ctx context.Context, asker suite.Subject, cands []siblin
// on a wake — the reader learns about it from the withholding notice on
// the page, which is where §6.5 puts it.
func (s *Siblings) Warm(ctx context.Context, store *Store, logger *slog.Logger) {
+ ctx, cancel := context.WithTimeout(ctx, siblingPollDeadline)
+ defer cancel()
var wg sync.WaitGroup
for _, app := range s.byName {
wg.Add(1)
diff --git a/internal/memoria/ingest_budget_test.go b/internal/memoria/ingest_budget_test.go
new file mode 100644
index 0000000..74d36a7
--- /dev/null
+++ b/internal/memoria/ingest_budget_test.go
@@ -0,0 +1,77 @@
+package memoria
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "net/http"
+ "net/http/httptest"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "amadan.net/oficina/memoria/internal/memoria/sibling"
+ "amadan.net/oficina/suite"
+ "gorm.io/gorm"
+)
+
+// Mutation-tested 2026-09-07: removing Poll's timeout failed at the real
+// cursor query, whose context no longer had a deadline.
+func TestPollEstablishesFiniteBudgetBeforeReadingItsCursor(t *testing.T) {
+ st := storeForTest(t)
+ var observed atomic.Bool
+ if err := st.db.G.Callback().Query().Before("gorm:query").Register("test:poll_deadline", func(tx *gorm.DB) {
+ if observed.Swap(true) {
+ return
+ }
+ d, ok := tx.Statement.Context.Deadline()
+ if !ok || time.Until(d) > 120*time.Second || time.Until(d) < 119*time.Second {
+ t.Errorf("poll deadline present=%v remaining=%v", ok, time.Until(d))
+ }
+ }); err != nil {
+ t.Fatal(err)
+ }
+ f := &feedServer{t: t, pages: map[string]suite.Feed{"": {Cursor: "end", Rows: []suite.FeedRow{}}, "end": {Cursor: "end", Rows: []suite.FeedRow{}}}}
+ sib, client := f.start(t)
+ if _, err := st.Poll(context.Background(), client, sib); err != nil {
+ t.Fatal(err)
+ }
+ if !observed.Load() {
+ t.Fatal("cursor query was not observed")
+ }
+}
+
+func TestPollDeadlineCoversSuccessiveContentRequests(t *testing.T) {
+ st := storeForTest(t)
+ var calls atomic.Int64
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path == suite.RouteS1 {
+ _ = json.NewEncoder(w).Encode(suite.Feed{Cursor: "next", Rows: []suite.FeedRow{row("first", 1, false), row("second", 1, false)}})
+ return
+ }
+ calls.Add(1)
+ select {
+ case <-time.After(180 * time.Millisecond):
+ _, _ = w.Write([]byte("synthetic body"))
+ case <-r.Context().Done():
+ }
+ }))
+ defer srv.Close()
+ sib := sibling.Sibling{Name: "Docs", BaseURL: srv.URL, Token: "synthetic"}
+ ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
+ defer cancel()
+ res, err := st.Poll(ctx, sibling.NewForTest(), sib)
+ if !errors.Is(err, context.DeadlineExceeded) {
+ t.Fatalf("poll deadline error=%v", err)
+ }
+ if calls.Load() != 2 || res.Stored != 1 {
+ t.Fatalf("requests=%d stored=%d", calls.Load(), res.Stored)
+ }
+ cursor, err := st.cursor(context.Background(), sib.Name)
+ if err != nil || cursor != "" {
+ t.Errorf("partial page advanced cursor: err=%v", err)
+ }
+ if !polledAt(t, st, sib.Name).IsZero() {
+ t.Error("partial page was marked current")
+ }
+}
diff --git a/internal/memoria/sibling/feed.go b/internal/memoria/sibling/feed.go
index a6589b5..fa0f490 100644
--- a/internal/memoria/sibling/feed.go
+++ b/internal/memoria/sibling/feed.go
@@ -7,10 +7,15 @@ import (
"io"
"net/http"
"net/url"
+ "time"
"amadan.net/oficina/suite"
)
+// Ingest allows the platform's 30s wake/header wait plus transfer margin.
+// This is a finite engineering bound; cold-wake latency still needs measurement.
+const ingestDeadline = 45 * time.Second
+
// maxFeedResponse and maxContentResponse cap what a sibling can make
// memoria hold in memory on one call.
//
@@ -40,7 +45,7 @@ const (
// comes back **only on success**: advancing past a page nobody read is how
// a consumer skips changes it will never be offered again.
func (c *Client) Feed(ctx context.Context, s Sibling, cursor string) (suite.Feed, error) {
- ctx, cancel := context.WithTimeout(ctx, Deadline)
+ ctx, cancel := context.WithTimeout(ctx, ingestDeadline)
defer cancel()
u := s.BaseURL + suite.RouteS1
@@ -88,7 +93,7 @@ func (c *Client) Feed(ctx context.Context, s Sibling, cursor string) (suite.Feed
// leaking, and folding that into ErrNotAvailable would hide the leak from
// the only person who could fix it.
func (c *Client) Content(ctx context.Context, s Sibling, itemID string, revision int64) ([]byte, error) {
- ctx, cancel := context.WithTimeout(ctx, Deadline)
+ ctx, cancel := context.WithTimeout(ctx, ingestDeadline)
defer cancel()
// **The item id is a PATH segment, not a query parameter.** The
diff --git a/internal/memoria/sibling/ingest_deadline_test.go b/internal/memoria/sibling/ingest_deadline_test.go
new file mode 100644
index 0000000..ef0186c
--- /dev/null
+++ b/internal/memoria/sibling/ingest_deadline_test.go
@@ -0,0 +1,154 @@
+package sibling
+
+import (
+ "context"
+ "errors"
+ "io"
+ "net/http"
+ "strings"
+ "testing"
+ "time"
+)
+
+func ingestCall(ctx context.Context, c *Client, s Sibling, content bool) error {
+ if content {
+ body, err := c.Content(ctx, s, "synthetic-item", 1)
+ if err == nil && string(body) != "synthetic body" {
+ return errors.New("unexpected body")
+ }
+ return err
+ }
+ feed, err := c.Feed(ctx, s, "synthetic-cursor")
+ if err == nil && (feed.Cursor != "next" || len(feed.Rows) != 0) {
+ return errors.New("unexpected feed")
+ }
+ return err
+}
+
+func ingestResponse(content bool) string {
+ if content {
+ return "synthetic body"
+ }
+ return `{"cursor":"next","rows":[]}`
+}
+
+// Mutation-tested 2026-09-07: recoupling Feed and Content separately to
+// the 800ms Deadline made the corresponding real HTTP case fail.
+func TestIngestCanFinishAfterTheSearchDeadline(t *testing.T) {
+ for _, content := range []bool{false, true} {
+ name := "S1"
+ if content {
+ name = "S3"
+ }
+ t.Run(name, func(t *testing.T) {
+ s := serve(t, func(w http.ResponseWriter, r *http.Request) {
+ timer := time.NewTimer(1100 * time.Millisecond)
+ defer timer.Stop()
+ select {
+ case <-timer.C:
+ _, _ = w.Write([]byte(ingestResponse(content)))
+ case <-r.Context().Done():
+ }
+ })
+ ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+ defer cancel()
+ if err := ingestCall(ctx, newTestClient(), s, content); err != nil {
+ t.Fatalf("valid ingest response within caller budget failed: %v", err)
+ }
+ })
+ }
+}
+
+// Mutation-tested 2026-09-07: discarding the request context failed all
+// four cases, both in the client and in the server cancellation observer.
+func TestIngestRetainsCallerCancellation(t *testing.T) {
+ for _, content := range []bool{false, true} {
+ for _, bodyStarted := range []bool{false, true} {
+ name := "S1"
+ if content {
+ name = "S3"
+ }
+ if bodyStarted {
+ name += "-body"
+ } else {
+ name += "-headers"
+ }
+ t.Run(name, func(t *testing.T) {
+ started := make(chan struct{})
+ observed := make(chan struct{})
+ s := serve(t, func(w http.ResponseWriter, r *http.Request) {
+ if bodyStarted {
+ _, _ = w.Write([]byte("{"))
+ w.(http.Flusher).Flush()
+ }
+ close(started)
+ select {
+ case <-r.Context().Done():
+ close(observed)
+ case <-time.After(2 * time.Second):
+ }
+ })
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ done := make(chan error, 1)
+ go func() { done <- ingestCall(ctx, newTestClient(), s, content) }()
+ select {
+ case <-started:
+ case <-time.After(time.Second):
+ t.Fatal("request did not start")
+ }
+ cancel()
+ select {
+ case err := <-done:
+ if !errors.Is(err, context.Canceled) {
+ t.Errorf("cancellation returned %v", err)
+ }
+ if err != nil && (strings.Contains(err.Error(), s.BaseURL) || strings.Contains(err.Error(), "synthetic-item") || strings.Contains(err.Error(), "synthetic-cursor")) {
+ t.Error("error disclosed request coordinates")
+ }
+ case <-time.After(500 * time.Millisecond):
+ t.Error("ingest ignored caller cancellation")
+ }
+ select {
+ case <-observed:
+ case <-time.After(500 * time.Millisecond):
+ t.Error("server did not observe cancellation")
+ }
+ })
+ }
+ }
+}
+
+func TestIngestRequestBudgetIsFiniteAndRespectsEarlierDeadline(t *testing.T) {
+ for _, content := range []bool{false, true} {
+ for _, earlier := range []bool{false, true} {
+ ctx := context.Background()
+ want := 45 * time.Second
+ cancel := func() {}
+ if earlier {
+ want = 2 * time.Second
+ ctx, cancel = context.WithTimeout(ctx, want)
+ }
+ var requestContext context.Context
+ client := &Client{http: &http.Client{Transport: ingestRoundTrip(func(r *http.Request) (*http.Response, error) {
+ requestContext = r.Context()
+ deadline, ok := requestContext.Deadline()
+ if !ok || time.Until(deadline) > want || time.Until(deadline) < want-time.Second {
+ t.Errorf("request budget: deadline present=%v remaining=%v want=%v", ok, time.Until(deadline), want)
+ }
+ return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(ingestResponse(content))), Header: make(http.Header)}, nil
+ })}}
+ if err := ingestCall(ctx, client, Sibling{Name: "Docs", BaseURL: "https://test.docs.oficina.works"}, content); err != nil {
+ t.Fatal(err)
+ }
+ if requestContext.Err() != context.Canceled {
+ t.Error("completed request retained its timeout context")
+ }
+ cancel()
+ }
+ }
+}
+
+type ingestRoundTrip func(*http.Request) (*http.Response, error)
+
+func (f ingestRoundTrip) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }
diff --git a/internal/memoria/siblingsync.go b/internal/memoria/siblingsync.go
index f2833d6..1b7af1e 100644
--- a/internal/memoria/siblingsync.go
+++ b/internal/memoria/siblingsync.go
@@ -23,6 +23,9 @@ import (
// stopped looking" is the difference this whole file is about.
const maxFeedPages = 50
+// One wake cannot spend a fresh request budget on every row indefinitely.
+const siblingPollDeadline = 2 * time.Minute
+
// PollResult is what one poll did. Counts only: no item id, no title, no
// body, so it can go in a log line or on the sources page (AGENTS.md).
type PollResult struct {
@@ -72,6 +75,8 @@ type PollResult struct {
// no-op. The expensive failure is skipping a row nobody will mention
// again.
func (s *Store) Poll(ctx context.Context, c *sibling.Client, sib sibling.Sibling) (PollResult, error) {
+ ctx, cancel := context.WithTimeout(ctx, siblingPollDeadline)
+ defer cancel()
var res PollResult
cursor, err := s.cursor(ctx, sib.Name)
if err != nil {