From 8ee5f984261a86ee67b92f2f3b61ee31b696ff01 Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Mon, 5 Oct 2026 17:33:28 +0000 Subject: [PATCH 01/15] feat: filter feedback records by updated_at for incremental extraction (ENG-3420) Add updated_since/updated_until to GET /v1/feedback-records and /count, backed by a (tenant_id, updated_at) index so an incremental sync does not scan the whole tenant. PATCH now stamps updated_at from the database clock like every enrichment write, so the column has one clock to compare against. Document the sync semantics: sort by an immutable column, derive the next watermark from returned updated_at minus an overlap, and reconcile deletions with a periodic full id sweep, since a deleted record has no row to match. --- .../core-concepts/filtering-and-sorting.mdx | 64 +++++++++- internal/api/validation/validation_test.go | 4 +- internal/models/feedback_records_filters.go | 14 ++- .../models/feedback_records_filters_test.go | 2 + .../repository/feedback_records_repository.go | 13 +- .../feedback_records_repository_test.go | 53 ++++++-- internal/repository/filter_conditions.go | 4 +- .../027_feedback_records_updated_at_index.sql | 31 +++++ openapi.yaml | 29 +++++ tests/feedback_record_filters_test.go | 116 ++++++++++++++++++ 10 files changed, 308 insertions(+), 22 deletions(-) create mode 100644 migrations/027_feedback_records_updated_at_index.sql diff --git a/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx b/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx index 959da92b..5fbf7868 100644 --- a/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx +++ b/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx @@ -81,13 +81,14 @@ narrow with the filters and post-process. ## Ranges -`collected_at`, `created_at`, `value_number`, `value_date` and `sentiment_score` take bounds. Every -bound is **inclusive** — a record sitting exactly on it is included. +`collected_at`, `created_at`, `updated_at`, `value_number`, `value_date` and `sentiment_score` take +bounds. Every bound is **inclusive** — a record sitting exactly on it is included. | Filter | Bounds | | --- | --- | | `collected_at` — when the feedback was given | `since` / `until` | | `created_at` — when Hub stored it | `created_since` / `created_until` | +| `updated_at` — when Hub last changed it | `updated_since` / `updated_until` | | `value_number` | `value_number_min` / `value_number_max` | | `value_date` | `value_date_min` / `value_date_max` | | `sentiment_score` | `sentiment_score_min` / `sentiment_score_max` | @@ -116,6 +117,65 @@ The two diverge whenever historical data is imported: the feedback was *collecte *created* in Hub today. Use `created_since` to answer "what did this import bring in", and `since` to answer "what did people tell us in this period". +## Incremental extraction + +To keep a copy of a tenant's records in sync without re-reading everything, filter on +`updated_since`. Hub sets `updated_at` when it stores a record and moves it forward whenever the +record changes, so one filter picks up both new and changed records: + +| Changes `updated_at` | Does not | +| --- | --- | +| creating the record | embeddings (stored apart from the record) | +| every `PATCH` | taxonomy assignment (not part of the listed record) | +| every translation, sentiment and emotion result | deleting the record — see below | + +Every writer stamps `updated_at` from the database clock, so the values are comparable with each +other and with the `updated_at` you read back. + +```bash +# everything created or changed since the last run, oldest first +curl -G "$HUB/v1/feedback-records" \ + -H "Authorization: Bearer $HUB_API_KEY" \ + -d tenant_id=org-123 \ + -d updated_since=2026-10-05T07:55:00Z \ + -d sort=created_at \ + -d order=asc \ + -d limit=1000 +``` + +**Sort by `created_at` or `collected_at`, not by `updated_at`.** `updated_at` is not a sort option, +on purpose: it changes under you, so a record could jump past the cursor and be skipped. The sort +columns never change, so every page stays stable even while records in the result are being +updated. A record that changes after you have paged past it now has a newer `updated_at`, and the +next run picks it up. + +### Choosing the next `updated_since` + +Take the **largest `updated_at` you received, minus an overlap**, and store it as the next run's +`updated_since`. If a run returns nothing, keep the previous value. + +- Use a value Hub gave you, not your own clock. It is the same clock Hub compares against, so clock + drift between your system and Hub cannot open a gap. +- Keep an overlap of a few minutes. `updated_at` is the moment the writing transaction *started*, + so a write that commits a moment after your read can carry a slightly earlier timestamp than + records you already saw. The overlap re-reads that window. +- The overlap means you will see some records twice. Upsert by `id`, and keep whichever copy has the + later `updated_at`. + +### Deleted records + + + +To reconcile deletions, periodically list **every** record id for the tenant: page through +`GET /v1/feedback-records` with no `updated_since` and a stable sort, such as `sort=created_at`. +Then remove any local record whose id was not returned. This catches every way a record can +disappear, including deletes by user and dataset purges. It reads the whole tenant, so run it less +often than the incremental sync, for example daily against an hourly sync. + ## Enrichment filters `sentiment`, `emotions` and `sentiment_score` filter on diff --git a/internal/api/validation/validation_test.go b/internal/api/validation/validation_test.go index 1328314a..a1d9bb48 100644 --- a/internal/api/validation/validation_test.go +++ b/internal/api/validation/validation_test.go @@ -226,6 +226,8 @@ func TestValidateInvertedRangeFilters(t *testing.T) { Until: &early, CreatedSince: &late, CreatedUntil: &early, + UpdatedSince: &late, + UpdatedUntil: &early, } err := ValidateStruct(filters) @@ -239,7 +241,7 @@ func TestValidateInvertedRangeFilters(t *testing.T) { names = append(names, FieldPath(fieldErr)) } - assert.ElementsMatch(t, []string{"since", "created_since"}, names) + assert.ElementsMatch(t, []string{"since", "created_since", "updated_since"}, names) }) // A lower bound with no counterpart is unbounded above, not inverted. Expressing this rule as diff --git a/internal/models/feedback_records_filters.go b/internal/models/feedback_records_filters.go index e9505830..3798c221 100644 --- a/internal/models/feedback_records_filters.go +++ b/internal/models/feedback_records_filters.go @@ -15,8 +15,9 @@ import ( // - immutable, because a mutable sort key lets a row move across the cursor between pages and be // silently skipped. This is why updated_at is deliberately absent: the enrichment workers // (SetTranslation, SetSentiment, writeEmotions) and every PATCH bump it, so `sort=updated_at` -// would lose rows under ordinary traffic. A change-feed needs an append-only sequence, not a -// sort parameter. +// would lose rows under ordinary traffic. Incremental extraction FILTERS on it instead +// (updated_since, ENG-3420) while ordering by an immutable column, which keeps every page +// stable however often the matching rows change. // // Adding a member requires a case in the repository's resolveListOrdering (token -> SQL column) // and in FeedbackRecord.SortValue (token -> record field). Both switches are exhaustive-linted, so @@ -98,6 +99,12 @@ type ListFeedbackRecordsFilters struct { CreatedSince *time.Time `form:"created_since" validate:"omitempty"` CreatedUntil *time.Time `form:"created_until" validate:"omitempty"` + // updated_at bounds, for incremental extraction (ENG-3420). Insert sets updated_at, and every + // PATCH and enrichment write bumps it, so updated_since selects records created OR changed + // since a point. It never surfaces a deleted record: a hard delete leaves no row to match. + UpdatedSince *time.Time `form:"updated_since" validate:"omitempty"` + UpdatedUntil *time.Time `form:"updated_until" validate:"omitempty"` + // Inclusive value bounds. Like every range filter here, these exclude rows whose column is // NULL — value_number_min=0 selects only records that carry a number at all. ValueNumberMin *float64 `form:"value_number_min" validate:"omitempty"` @@ -155,10 +162,11 @@ type InvertedRangeFilter struct { // the check, which TestInvertedRanges_CoversEveryRangeFilter guards against by reflecting over the // struct's form tags. func (f *ListFeedbackRecordsFilters) InvertedRanges() []InvertedRangeFilter { - inverted := make([]InvertedRangeFilter, 0, 5) //nolint:mnd // capacity hint: the five pairs below + inverted := make([]InvertedRangeFilter, 0, 6) //nolint:mnd // capacity hint: the six pairs below appendInvertedTimeRange(&inverted, f.Since, f.Until, "Since", "since", "until") appendInvertedTimeRange(&inverted, f.CreatedSince, f.CreatedUntil, "CreatedSince", "created_since", "created_until") + appendInvertedTimeRange(&inverted, f.UpdatedSince, f.UpdatedUntil, "UpdatedSince", "updated_since", "updated_until") appendInvertedTimeRange(&inverted, f.ValueDateMin, f.ValueDateMax, "ValueDateMin", "value_date_min", "value_date_max") appendInvertedFloatRange( &inverted, f.ValueNumberMin, f.ValueNumberMax, "ValueNumberMin", "value_number_min", "value_number_max", diff --git a/internal/models/feedback_records_filters_test.go b/internal/models/feedback_records_filters_test.go index 524180b3..17316289 100644 --- a/internal/models/feedback_records_filters_test.go +++ b/internal/models/feedback_records_filters_test.go @@ -204,6 +204,7 @@ func TestInvertedRanges_ReportsEachInvertedPair(t *testing.T) { want := map[string]string{ "since": "until", "created_since": "created_until", + "updated_since": "updated_until", "value_date_min": "value_date_max", "value_number_min": "value_number_max", "sentiment_score_min": "sentiment_score_max", @@ -280,6 +281,7 @@ func invertedFilters() *ListFeedbackRecordsFilters { return &ListFeedbackRecordsFilters{ Since: &late, Until: &early, CreatedSince: &late, CreatedUntil: &early, + UpdatedSince: &late, UpdatedUntil: &early, ValueDateMin: &late, ValueDateMax: &early, ValueNumberMin: &high, ValueNumberMax: &low, SentimentScoreMin: &high, SentimentScoreMax: &low, diff --git a/internal/repository/feedback_records_repository.go b/internal/repository/feedback_records_repository.go index 0b7a0789..dded55d9 100644 --- a/internal/repository/feedback_records_repository.go +++ b/internal/repository/feedback_records_repository.go @@ -790,8 +790,13 @@ func (r *FeedbackRecordsRepository) ListAfterCursor( // buildUpdateQuery builds an UPDATE query with SET clause and arguments. // Returns the query string, arguments, and a boolean indicating if any updates were provided. +// +// updated_at is stamped from the database clock (NOW()), the same clock every enrichment write and +// the insert default use. updated_since (ENG-3420) compares that column against a caller's +// watermark, so one writer on a pod's clock would let that pod's skew hide its edits from an +// incremental extraction. func buildUpdateQuery( - req *models.UpdateFeedbackRecordRequest, id uuid.UUID, updatedAt time.Time, + req *models.UpdateFeedbackRecordRequest, id uuid.UUID, ) (query string, args []any, hasUpdates bool) { var updates []string @@ -910,9 +915,7 @@ func buildUpdateQuery( return "", nil, false } - updates = append(updates, fmt.Sprintf("updated_at = $%d", argCount)) - args = append(args, updatedAt) - argCount++ + updates = append(updates, "updated_at = NOW()") args = append(args, id) @@ -956,7 +959,7 @@ func joinOr(conds ...string) string { func (r *FeedbackRecordsRepository) Update( ctx context.Context, id uuid.UUID, req *models.UpdateFeedbackRecordRequest, ) (updated, previous *models.FeedbackRecord, err error) { - query, args, hasUpdates := buildUpdateQuery(req, id, time.Now()) + query, args, hasUpdates := buildUpdateQuery(req, id) if !hasUpdates { // No write happens, so no tenant write lock is needed; nothing changed, so the previous // state is the current row. diff --git a/internal/repository/feedback_records_repository_test.go b/internal/repository/feedback_records_repository_test.go index d30ce38f..1d0b02c0 100644 --- a/internal/repository/feedback_records_repository_test.go +++ b/internal/repository/feedback_records_repository_test.go @@ -72,7 +72,7 @@ func TestBuildUpdateQuery_ClearsStaleEnrichmentOnContentChange(t *testing.T) { for _, testCase := range cases { t.Run(testCase.name, func(t *testing.T) { - query, _, hasUpdates := buildUpdateQuery(testCase.req, uuid.New(), time.Now()) + query, _, hasUpdates := buildUpdateQuery(testCase.req, uuid.New()) if !hasUpdates { t.Fatal("buildUpdateQuery hasUpdates = false, want true") } @@ -200,7 +200,7 @@ func TestBuildUpdateQuery_ValueID(t *testing.T) { valueID := "opt_very_satisfied" req := &models.UpdateFeedbackRecordRequest{ValueID: &valueID} - query, args, hasUpdates := buildUpdateQuery(req, uuid.New(), time.Now()) + query, args, hasUpdates := buildUpdateQuery(req, uuid.New()) if !hasUpdates { t.Fatal("buildUpdateQuery hasUpdates = false, want true") } @@ -218,6 +218,34 @@ func TestBuildUpdateQuery_ValueID(t *testing.T) { } } +// TestBuildUpdateQuery_StampsUpdatedAtFromDatabaseClock pins PATCH to the database clock, the one +// every enrichment write and the insert default use. updated_since compares updated_at against a +// caller's watermark (ENG-3420), so a timestamp bound from the pod's clock would let that pod's +// skew hide an edit from an incremental extraction. +func TestBuildUpdateQuery_StampsUpdatedAtFromDatabaseClock(t *testing.T) { + userID := "u1" + id := uuid.New() + + query, args, hasUpdates := buildUpdateQuery(&models.UpdateFeedbackRecordRequest{UserID: &userID}, id) + if !hasUpdates { + t.Fatal("buildUpdateQuery hasUpdates = false, want true") + } + + if !strings.Contains(query, "updated_at = NOW()") { + t.Fatalf("query must stamp updated_at from the database clock\nquery: %s", query) + } + + for i, arg := range args { + if _, isTime := arg.(time.Time); isTime { + t.Fatalf("args[%d] binds a time.Time; updated_at must not come from the caller's clock", i) + } + } + + if !reflect.DeepEqual(args, []any{userID, id}) { + t.Fatalf("args = %v, want [user_id, id]", args) + } +} + // TestBuildFilterConditions_PlaceholdersMatchArgs locks that every generated $N placeholder maps to // its argument's 1-based position for any combination of filters. The placeholder is derived from // len(args)+1 at each append precisely so the order of filters can't desync it — this guards @@ -230,6 +258,8 @@ func TestBuildFilterConditions_PlaceholdersMatchArgs(t *testing.T) { until := time.Date(2026, 2, 1, 0, 0, 0, 0, time.UTC) createdSince := time.Date(2026, 3, 1, 0, 0, 0, 0, time.UTC) createdUntil := time.Date(2026, 4, 1, 0, 0, 0, 0, time.UTC) + updatedSince := time.Date(2026, 4, 15, 0, 0, 0, 0, time.UTC) + updatedUntil := time.Date(2026, 4, 30, 0, 0, 0, 0, time.UTC) valueDateMin := time.Date(2026, 5, 1, 0, 0, 0, 0, time.UTC) valueDateMax := time.Date(2026, 6, 1, 0, 0, 0, 0, time.UTC) numberMin, numberMax := 1.5, 9.5 @@ -251,6 +281,7 @@ func TestBuildFilterConditions_PlaceholdersMatchArgs(t *testing.T) { Since: &since, Until: &until, CreatedSince: &createdSince, CreatedUntil: &createdUntil, + UpdatedSince: &updatedSince, UpdatedUntil: &updatedUntil, ValueDateMin: &valueDateMin, ValueDateMax: &valueDateMax, ValueNumberMin: &numberMin, ValueNumberMax: &numberMax, @@ -284,14 +315,16 @@ func TestBuildFilterConditions_PlaceholdersMatchArgs(t *testing.T) { {"collected_at <= $13", until}, {"created_at >= $14", createdSince}, {"created_at <= $15", createdUntil}, - {"value_date >= $16", valueDateMin}, - {"value_date <= $17", valueDateMax}, - {"value_number >= $18", numberMin}, - {"value_number <= $19", numberMax}, - {"sentiment = ANY($20)", []string{"negative", "very_negative"}}, - {"sentiment_score >= $21", scoreMin}, - {"sentiment_score <= $22", scoreMax}, - {"emotions && $23", []string{"anger"}}, + {"updated_at >= $16", updatedSince}, + {"updated_at <= $17", updatedUntil}, + {"value_date >= $18", valueDateMin}, + {"value_date <= $19", valueDateMax}, + {"value_number >= $20", numberMin}, + {"value_number <= $21", numberMax}, + {"sentiment = ANY($22)", []string{"negative", "very_negative"}}, + {"sentiment_score >= $23", scoreMin}, + {"sentiment_score <= $24", scoreMax}, + {"emotions && $25", []string{"anger"}}, } if len(args) != len(expected) { diff --git a/internal/repository/filter_conditions.go b/internal/repository/filter_conditions.go index d37708c1..5b1e2792 100644 --- a/internal/repository/filter_conditions.go +++ b/internal/repository/filter_conditions.go @@ -32,6 +32,7 @@ const ( colLanguage sqlColumn = "language" colCollectedAt sqlColumn = "collected_at" colCreatedAt sqlColumn = "created_at" + colUpdatedAt sqlColumn = "updated_at" colValueDate sqlColumn = "value_date" colValueNumber sqlColumn = "value_number" colSentiment sqlColumn = "sentiment" @@ -195,10 +196,11 @@ func appendIdentityConditions(conds *filterConditions, filters *models.ListFeedb // appendTimeConditions adds the inclusive timestamp ranges. collected_at is when the feedback was // given; created_at is when Hub stored it. They diverge on a historical re-import, which is why -// each has its own pair. +// each has its own pair. updated_at is when Hub last wrote the row, for incremental extraction. func appendTimeConditions(conds *filterConditions, filters *models.ListFeedbackRecordsFilters) { addTimeRange(conds, colCollectedAt, filters.Since, filters.Until) addTimeRange(conds, colCreatedAt, filters.CreatedSince, filters.CreatedUntil) + addTimeRange(conds, colUpdatedAt, filters.UpdatedSince, filters.UpdatedUntil) addTimeRange(conds, colValueDate, filters.ValueDateMin, filters.ValueDateMax) } diff --git a/migrations/027_feedback_records_updated_at_index.sql b/migrations/027_feedback_records_updated_at_index.sql new file mode 100644 index 00000000..9d9e8f9b --- /dev/null +++ b/migrations/027_feedback_records_updated_at_index.sql @@ -0,0 +1,31 @@ +-- +goose NO TRANSACTION +-- +goose up +-- Index support for the updated_since / updated_until filters on GET /v1/feedback-records and +-- .../count (ENG-3420). +-- +-- updated_since exists for incremental extraction: "every record created or changed since my last +-- run". That query is narrow by design — a sync that runs hourly matches a sliver of the tenant — +-- and without an index the planner can only satisfy it by walking +-- idx_feedback_records_tenant_collected_at_id and discarding rows, so every page of every sync +-- reads the whole tenant. That is the degradation migration 021 indexed created_at against, and it +-- gets worse exactly as a tenant grows large enough to need incremental extraction at all. +-- +-- (tenant_id, updated_at), with no id and no DESC: unlike created_at, updated_at is never a sort +-- key (it is mutable, see models.SortField), so this index only has to find the matching rows. The +-- listing still orders by collected_at or created_at; a narrow match set is fetched here and +-- sorted, and a wide one falls back to walking the ordering index, which the planner picks by cost. +-- +-- Write cost, accepted deliberately: updated_at changes on every PATCH and every enrichment write, +-- so each of those updates now maintains this index too. Sentiment, emotions and PATCH writes +-- already touch indexed columns and were not HOT updates anyway; a translation write touched none +-- and could be HOT until now. One extra index entry per enrichment write is the price of an +-- extraction query that does not scan the tenant. +-- +-- Runs without a transaction because of CONCURRENTLY, and is re-runnable: DROP-then-CREATE replaces +-- the INVALID index an interrupted CREATE INDEX CONCURRENTLY leaves behind (see 021). +DROP INDEX CONCURRENTLY IF EXISTS idx_feedback_records_tenant_updated_at; +CREATE INDEX CONCURRENTLY idx_feedback_records_tenant_updated_at + ON feedback_records (tenant_id, updated_at); + +-- +goose down +DROP INDEX CONCURRENTLY IF EXISTS idx_feedback_records_tenant_updated_at; diff --git a/openapi.yaml b/openapi.yaml index c3c2b5f3..9c8adfbc 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -98,6 +98,8 @@ paths: - $ref: '#/components/parameters/FeedbackRecordsUntil' - $ref: '#/components/parameters/FeedbackRecordsCreatedSince' - $ref: '#/components/parameters/FeedbackRecordsCreatedUntil' + - $ref: '#/components/parameters/FeedbackRecordsUpdatedSince' + - $ref: '#/components/parameters/FeedbackRecordsUpdatedUntil' - $ref: '#/components/parameters/FeedbackRecordsValueNumberMin' - $ref: '#/components/parameters/FeedbackRecordsValueNumberMax' - $ref: '#/components/parameters/FeedbackRecordsValueDateMin' @@ -512,6 +514,8 @@ paths: - $ref: '#/components/parameters/FeedbackRecordsUntil' - $ref: '#/components/parameters/FeedbackRecordsCreatedSince' - $ref: '#/components/parameters/FeedbackRecordsCreatedUntil' + - $ref: '#/components/parameters/FeedbackRecordsUpdatedSince' + - $ref: '#/components/parameters/FeedbackRecordsUpdatedUntil' - $ref: '#/components/parameters/FeedbackRecordsValueNumberMin' - $ref: '#/components/parameters/FeedbackRecordsValueNumberMax' - $ref: '#/components/parameters/FeedbackRecordsValueDateMin' @@ -2719,6 +2723,31 @@ components: type: string format: date-time example: "2026-12-31T00:00:00Z" + FeedbackRecordsUpdatedSince: + name: updated_since + in: query + description: | + Filter by updated_at >= updated_since (ISO 8601, inclusive). updated_at is set when Hub + stores a record and bumped by every update and every enrichment result (translation, + sentiment, emotions), so this selects records created OR changed since a point — the + parameter for incremental extraction. Permanently deleted records are never returned: + reconcile deletions separately by listing record ids. updated_at is not a sort key; + keep sorting by collected_at or created_at, which stays stable while matching records + change. + schema: + type: string + format: date-time + example: "2026-01-01T00:00:00Z" + FeedbackRecordsUpdatedUntil: + name: updated_until + in: query + description: | + Filter by updated_at <= updated_until (ISO 8601, inclusive). See updated_since for what + updates updated_at. + schema: + type: string + format: date-time + example: "2026-12-31T00:00:00Z" FeedbackRecordsValueNumberMin: name: value_number_min in: query diff --git a/tests/feedback_record_filters_test.go b/tests/feedback_record_filters_test.go index e102074c..cc63dc4e 100644 --- a/tests/feedback_record_filters_test.go +++ b/tests/feedback_record_filters_test.go @@ -282,6 +282,121 @@ func TestFeedbackRecordFilters_CreatedAtIsDistinctFromCollectedAt(t *testing.T) "created_since bounds created_at, which is after the cutoff") } +// dbNow reads the database clock, the one every writer stamps updated_at from. A cutoff taken from +// the test process's clock would make these tests depend on the two clocks agreeing. +func (e *filterTestEnv) dbNow(t *testing.T) time.Time { + t.Helper() + + var now time.Time + require.NoError(t, e.db.QueryRow(t.Context(), `SELECT clock_timestamp()`).Scan(&now)) + + return now +} + +// TestFeedbackRecordFilters_UpdatedSinceSelectsCreatedAndChanged drives each real write path that +// should move updated_at — insert, PATCH, an enrichment result — and checks updated_since selects +// exactly those records, while an untouched older record stays out (ENG-3420). +func TestFeedbackRecordFilters_UpdatedSinceSelectsCreatedAndChanged(t *testing.T) { + env := newFilterTestEnv(t, "updated-since") + + untouched := env.seed(t) + patched := env.seed(t) + enriched := env.seed(t) + + cutoff := env.dbNow(t) + + updated, _, err := env.repo.Update(t.Context(), patched.ID, &models.UpdateFeedbackRecordRequest{ + UserID: new("user-after-cutoff"), + }) + require.NoError(t, err) + require.False(t, updated.UpdatedAt.Before(cutoff), "PATCH must move updated_at past the cutoff") + + sentiment, score := models.SentimentPositive, 0.5 + require.NoError(t, env.repo.SetSentiment(t.Context(), enriched.ID, &sentiment, &score, nil)) + + created := env.seed(t) + + since := func(f *models.ListFeedbackRecordsFilters) { f.UpdatedSince = &cutoff } + assert.ElementsMatch(t, []uuid.UUID{patched.ID, enriched.ID, created.ID}, env.list(t, since), + "updated_since selects records created or changed at or after the cutoff") + assert.Equal(t, 3, env.count(t, since), "count must describe the same set") + + until := func(f *models.ListFeedbackRecordsFilters) { f.UpdatedUntil = &cutoff } + assert.Equal(t, []uuid.UUID{untouched.ID}, env.list(t, until), + "updated_until bounds updated_at, so only the record nobody touched is at or before the cutoff") +} + +// TestFeedbackRecordFilters_UpdatedSincePaginationSurvivesUpdates is the reason updated_at is a +// filter and never a sort key. Records keep changing while a sync pages through them; ordering by +// an immutable column means an update moves a record's updated_at but never its position, so the +// traversal still returns every record exactly once. +func TestFeedbackRecordFilters_UpdatedSincePaginationSurvivesUpdates(t *testing.T) { + env := newFilterTestEnv(t, "updated-since-pages") + svc := service.NewFeedbackRecordsService( + repository.NewFeedbackRecordsRepository(env.db), nil, "", nil, nil, "", 0, "", + ) + + cutoff := env.dbNow(t) + + const records, pageLimit = 6, 2 + + want := make(map[uuid.UUID]bool, records) + ordered := make([]uuid.UUID, 0, records) + + for range records { + record := env.seed(t) + want[record.ID] = true + ordered = append(ordered, record.ID) + } + + touch := func(id uuid.UUID) { + _, _, err := env.repo.Update(t.Context(), id, &models.UpdateFeedbackRecordRequest{ + UserID: new("touched-" + uuid.NewString()), + }) + require.NoError(t, err) + } + + seen := make(map[uuid.UUID]int, records) + nextCursor := "" + + for page := range records { + tenant := env.tenant + resp, err := svc.ListFeedbackRecords(t.Context(), &models.ListFeedbackRecordsFilters{ + TenantID: &tenant, + UpdatedSince: &cutoff, + Sort: models.SortFieldCreatedAt, + Order: models.SortOrderAsc, + Limit: pageLimit, + Cursor: nextCursor, + }) + require.NoError(t, err, "page %d", page) + + for _, record := range resp.Data { + seen[record.ID]++ + } + + if page == 0 { + // Change one record already returned and one not reached yet. Under sort=updated_at + // the second would jump behind the cursor and be skipped. + touch(ordered[0]) + touch(ordered[records-1]) + } + + nextCursor = resp.NextCursor + if nextCursor == "" { + break + } + } + + require.Empty(t, nextCursor, "pagination did not terminate") + assert.Len(t, seen, records, "every record must be returned") + + for id, times := range seen { + assert.True(t, want[id], "record %s does not belong to this traversal", id) + assert.Equal(t, 1, times, "record %s returned %d times", id, times) + } +} + // TestFeedbackRecordFilters_PresencePartitions verifies each presence filter reads the column it // is supposed to, and that true/false partition the tenant exactly. // @@ -435,6 +550,7 @@ func TestFeedbackRecordFilters_TenantIsolation(t *testing.T) { {"has_sentiment", func(f *models.ListFeedbackRecordsFilters) { f.HasSentiment = new(true) }}, {"collected_at range", func(f *models.ListFeedbackRecordsFilters) { f.Since = &collected }}, {"created_at range", func(f *models.ListFeedbackRecordsFilters) { f.CreatedSince = &collected }}, + {"updated_at range", func(f *models.ListFeedbackRecordsFilters) { f.UpdatedSince = &collected }}, } for _, tt := range tests { From 30c5da7aa56a231445a10d3d0f076f7d9bd552b3 Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Mon, 5 Oct 2026 17:44:44 +0000 Subject: [PATCH 02/15] docs: anchor the updated_since watermark to the run's start (ENG-3420) A watermark taken from the largest updated_at received skips a record that changes after a multi-page run has paged past it. Prescribe the run's start (the first response's Date header) minus an overlap, let the latest fetch win instead of comparing updated_at (not monotonic per row), and only reconcile deletions for records created before the sweep began. Pin the procedure with a two-run integration test, and cover the emotions and translation writers plus a pre-cutoff record. --- .../core-concepts/filtering-and-sorting.mdx | 55 ++++---- internal/models/feedback_records_filters.go | 8 +- openapi.yaml | 15 ++- tests/feedback_record_filters_test.go | 121 +++++++++++++++++- 4 files changed, 160 insertions(+), 39 deletions(-) diff --git a/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx b/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx index 5fbf7868..c44eb8c5 100644 --- a/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx +++ b/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx @@ -120,17 +120,16 @@ to answer "what did people tell us in this period". ## Incremental extraction To keep a copy of a tenant's records in sync without re-reading everything, filter on -`updated_since`. Hub sets `updated_at` when it stores a record and moves it forward whenever the -record changes, so one filter picks up both new and changed records: +`updated_since`. Hub stamps `updated_at` on every write to a record, so one filter picks up both +new and changed records: -| Changes `updated_at` | Does not | +| Stamps `updated_at` | Does not | | --- | --- | | creating the record | embeddings (stored apart from the record) | | every `PATCH` | taxonomy assignment (not part of the listed record) | | every translation, sentiment and emotion result | deleting the record — see below | -Every writer stamps `updated_at` from the database clock, so the values are comparable with each -other and with the `updated_at` you read back. +Every writer takes the timestamp from the database clock, at the moment its transaction starts. ```bash # everything created or changed since the last run, oldest first @@ -146,21 +145,28 @@ curl -G "$HUB/v1/feedback-records" \ **Sort by `created_at` or `collected_at`, not by `updated_at`.** `updated_at` is not a sort option, on purpose: it changes under you, so a record could jump past the cursor and be skipped. The sort columns never change, so every page stays stable even while records in the result are being -updated. A record that changes after you have paged past it now has a newer `updated_at`, and the -next run picks it up. +updated. ### Choosing the next `updated_since` -Take the **largest `updated_at` you received, minus an overlap**, and store it as the next run's -`updated_since`. If a run returns nothing, keep the previous value. - -- Use a value Hub gave you, not your own clock. It is the same clock Hub compares against, so clock - drift between your system and Hub cannot open a gap. -- Keep an overlap of a few minutes. `updated_at` is the moment the writing transaction *started*, - so a write that commits a moment after your read can carry a slightly earlier timestamp than - records you already saw. The overlap re-reads that window. -- The overlap means you will see some records twice. Upsert by `id`, and keep whichever copy has the - later `updated_at`. +**Note when the run starts, and use that time minus an overlap as the next run's `updated_since`.** +Do not use the largest `updated_at` you received. A run that spans several pages reads each page at +a different moment: a record you paged past early can change while later pages are still being +read, and its new `updated_at` is then older than the newest one on your last page. A watermark +taken from the results skips that change. A watermark taken from the run's start does not, because +anything written after the run began is newer than it. + +- **Take the start time from Hub, not from your own clock.** Every response carries an HTTP `Date` + header; the one on the run's first response is the run's start on Hub's side, so drift between + your clock and Hub's cannot open a gap. +- **Keep an overlap of a few minutes, five is a safe default.** `updated_at` is when the writing + transaction *started*, so a write that was still committing when your run began carries a + slightly earlier timestamp. The overlap has to cover the longest write plus the difference between + Hub's and its database's clocks. +- **The latest fetch wins.** The overlap re-reads some records, so upsert by `id` and let the copy + from the most recent run replace the older one. Do not compare `updated_at` to pick a copy: two + writes to one record can commit in the opposite order to the one they started in, so a newer state + can carry an older timestamp. A later read always sees the later state. ### Deleted records @@ -170,11 +176,16 @@ Take the **largest `updated_at` you received, minus an overlap**, and store it a that Hub has since deleted. -To reconcile deletions, periodically list **every** record id for the tenant: page through -`GET /v1/feedback-records` with no `updated_since` and a stable sort, such as `sort=created_at`. -Then remove any local record whose id was not returned. This catches every way a record can -disappear, including deletes by user and dataset purges. It reads the whole tenant, so run it less -often than the incremental sync, for example daily against an hourly sync. +To reconcile deletions, periodically list **every** record id for the tenant: note the run's start +from the first response's `Date` header, then page through `GET /v1/feedback-records` with no +`updated_since` and a stable sort, such as `sort=created_at`. Afterwards, remove the local records +that were not returned **and whose `created_at` is earlier than that start minus the overlap**. +That condition matters: a record created after the sweep began may have been stored by your +incremental sync but never reached by the sweep, and deleting it would lose it for good. + +This catches every way a record can disappear, including deletes by user and dataset purges. It +reads the whole tenant, so run it less often than the incremental sync, for example daily against +an hourly sync. ## Enrichment filters diff --git a/internal/models/feedback_records_filters.go b/internal/models/feedback_records_filters.go index 3798c221..7e276666 100644 --- a/internal/models/feedback_records_filters.go +++ b/internal/models/feedback_records_filters.go @@ -99,9 +99,11 @@ type ListFeedbackRecordsFilters struct { CreatedSince *time.Time `form:"created_since" validate:"omitempty"` CreatedUntil *time.Time `form:"created_until" validate:"omitempty"` - // updated_at bounds, for incremental extraction (ENG-3420). Insert sets updated_at, and every - // PATCH and enrichment write bumps it, so updated_since selects records created OR changed - // since a point. It never surfaces a deleted record: a hard delete leaves no row to match. + // updated_at bounds, for incremental extraction (ENG-3420). Insert, PATCH and every enrichment + // write stamp updated_at with their transaction's NOW(), so updated_since selects records created + // OR changed since a point. Not monotonic per row: two writes queued on one record's lock can + // commit in the opposite order to the one they started in. It never surfaces a deleted record: + // a hard delete leaves no row to match. UpdatedSince *time.Time `form:"updated_since" validate:"omitempty"` UpdatedUntil *time.Time `form:"updated_until" validate:"omitempty"` diff --git a/openapi.yaml b/openapi.yaml index 9c8adfbc..6e7e9be9 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -2727,13 +2727,14 @@ components: name: updated_since in: query description: | - Filter by updated_at >= updated_since (ISO 8601, inclusive). updated_at is set when Hub - stores a record and bumped by every update and every enrichment result (translation, - sentiment, emotions), so this selects records created OR changed since a point — the - parameter for incremental extraction. Permanently deleted records are never returned: - reconcile deletions separately by listing record ids. updated_at is not a sort key; - keep sorting by collected_at or created_at, which stays stable while matching records - change. + Filter by updated_at >= updated_since (ISO 8601, inclusive). updated_at is stamped on + every write to a record — create, update, and every enrichment result (translation, + sentiment, emotions) — so this selects records created OR changed since a point: the + parameter for incremental extraction. For the next run, use the time the previous run + started (its first response's Date header) minus a few minutes of overlap, not the + largest updated_at received. Permanently deleted records are never returned: reconcile + deletions separately by listing record ids. updated_at is not a sort key; keep sorting + by collected_at or created_at, which stays stable while matching records change. schema: type: string format: date-time diff --git a/tests/feedback_record_filters_test.go b/tests/feedback_record_filters_test.go index cc63dc4e..7443011c 100644 --- a/tests/feedback_record_filters_test.go +++ b/tests/feedback_record_filters_test.go @@ -293,15 +293,18 @@ func (e *filterTestEnv) dbNow(t *testing.T) time.Time { return now } -// TestFeedbackRecordFilters_UpdatedSinceSelectsCreatedAndChanged drives each real write path that -// should move updated_at — insert, PATCH, an enrichment result — and checks updated_since selects -// exactly those records, while an untouched older record stays out (ENG-3420). +// TestFeedbackRecordFilters_UpdatedSinceSelectsCreatedAndChanged drives each real write path the +// docs promise stamps updated_at — insert, PATCH, and the sentiment, emotions and translation +// results — and checks updated_since selects exactly those records, while an untouched older record +// stays out (ENG-3420). func TestFeedbackRecordFilters_UpdatedSinceSelectsCreatedAndChanged(t *testing.T) { env := newFilterTestEnv(t, "updated-since") untouched := env.seed(t) patched := env.seed(t) - enriched := env.seed(t) + withSentiment := env.seed(t) + withEmotions := env.seed(t) + translated := env.seed(t) cutoff := env.dbNow(t) @@ -312,14 +315,20 @@ func TestFeedbackRecordFilters_UpdatedSinceSelectsCreatedAndChanged(t *testing.T require.False(t, updated.UpdatedAt.Before(cutoff), "PATCH must move updated_at past the cutoff") sentiment, score := models.SentimentPositive, 0.5 - require.NoError(t, env.repo.SetSentiment(t.Context(), enriched.ID, &sentiment, &score, nil)) + require.NoError(t, env.repo.SetSentiment(t.Context(), withSentiment.ID, &sentiment, &score, nil)) + require.NoError(t, env.repo.SetEmotions(t.Context(), withEmotions.ID, []models.EmotionValue{models.EmotionJoy}, nil)) + + // No tenant settings row, so the effective target is the default passed here and the write lands. + translation := "etwas Feedback" + require.NoError(t, env.repo.SetTranslation(t.Context(), translated.ID, &translation, "de", "de", nil)) created := env.seed(t) + want := []uuid.UUID{patched.ID, withSentiment.ID, withEmotions.ID, translated.ID, created.ID} since := func(f *models.ListFeedbackRecordsFilters) { f.UpdatedSince = &cutoff } - assert.ElementsMatch(t, []uuid.UUID{patched.ID, enriched.ID, created.ID}, env.list(t, since), + assert.ElementsMatch(t, want, env.list(t, since), "updated_since selects records created or changed at or after the cutoff") - assert.Equal(t, 3, env.count(t, since), "count must describe the same set") + assert.Equal(t, len(want), env.count(t, since), "count must describe the same set") until := func(f *models.ListFeedbackRecordsFilters) { f.UpdatedUntil = &cutoff } assert.Equal(t, []uuid.UUID{untouched.ID}, env.list(t, until), @@ -336,6 +345,9 @@ func TestFeedbackRecordFilters_UpdatedSincePaginationSurvivesUpdates(t *testing. repository.NewFeedbackRecordsRepository(env.db), nil, "", nil, nil, "", 0, "", ) + // Written before the cutoff, so it must never appear: without it this test would pass with + // updated_since ignored altogether. + before := env.seed(t) cutoff := env.dbNow(t) const records, pageLimit = 6, 2 @@ -389,6 +401,7 @@ func TestFeedbackRecordFilters_UpdatedSincePaginationSurvivesUpdates(t *testing. } require.Empty(t, nextCursor, "pagination did not terminate") + assert.NotContains(t, seen, before.ID, "a record last written before updated_since must not be returned") assert.Len(t, seen, records, "every record must be returned") for id, times := range seen { @@ -397,6 +410,100 @@ func TestFeedbackRecordFilters_UpdatedSincePaginationSurvivesUpdates(t *testing. } } +// TestFeedbackRecordFilters_UpdatedSinceWatermarkFromRunStart drives the sync procedure the docs +// prescribe across two runs, in the one shape that makes the choice of watermark matter: a record +// changes after the run has paged past it, while a record further along changes later still. +// +// The newest updated_at the run received is then AFTER the first record's change, so a watermark +// taken from the results would skip that change for good. The watermark the docs prescribe — the +// run's start — is before it, and the second run returns it. +func TestFeedbackRecordFilters_UpdatedSinceWatermarkFromRunStart(t *testing.T) { + env := newFilterTestEnv(t, "updated-since-watermark") + svc := service.NewFeedbackRecordsService( + repository.NewFeedbackRecordsRepository(env.db), nil, "", nil, nil, "", 0, "", + ) + + first := env.seed(t) + env.seed(t) + last := env.seed(t) + + touch := func(id uuid.UUID) { + _, _, err := env.repo.Update(t.Context(), id, &models.UpdateFeedbackRecordRequest{ + UserID: new("touched-" + uuid.NewString()), + }) + require.NoError(t, err) + } + + // run pages through every record updated at or after since, one per page, calling between + // after the first page. It returns the records in the order received. + run := func(since time.Time, between func()) []*models.FeedbackRecord { + var ( + received []*models.FeedbackRecord + nextCursor string + ) + + for page := range 10 { + tenant := env.tenant + resp, err := svc.ListFeedbackRecords(t.Context(), &models.ListFeedbackRecordsFilters{ + TenantID: &tenant, UpdatedSince: &since, + Sort: models.SortFieldCreatedAt, Order: models.SortOrderAsc, + Limit: 1, Cursor: nextCursor, + }) + require.NoError(t, err) + + for i := range resp.Data { + received = append(received, &resp.Data[i]) + } + + if page == 0 && between != nil { + between() + } + + nextCursor = resp.NextCursor + if nextCursor == "" { + return received + } + } + + t.Fatal("pagination did not terminate") + + return nil + } + + runStart := env.dbNow(t) + firstRun := run(time.Time{}, func() { + touch(first.ID) // already paged past + touch(last.ID) // not reached yet, changed later still + }) + require.Len(t, firstRun, 3) + require.Equal(t, first.ID, firstRun[0].ID) + + newestReceived := firstRun[0].UpdatedAt + for _, record := range firstRun { + if record.UpdatedAt.After(newestReceived) { + newestReceived = record.UpdatedAt + } + } + + var firstChangedAt time.Time + require.NoError(t, env.db.QueryRow(t.Context(), + `SELECT updated_at FROM feedback_records WHERE id = $1`, first.ID).Scan(&firstChangedAt)) + + // The scenario is real: a results-derived watermark would already be past the first change. + require.True(t, firstChangedAt.Before(newestReceived), + "the first record's change (%v) must be older than the newest updated_at received (%v)", + firstChangedAt, newestReceived) + + secondRun := run(runStart, nil) + + ids := make([]uuid.UUID, 0, len(secondRun)) + for _, record := range secondRun { + ids = append(ids, record.ID) + } + + assert.Contains(t, ids, first.ID, "a run-start watermark must return the change made behind the cursor") +} + // TestFeedbackRecordFilters_PresencePartitions verifies each presence filter reads the column it // is supposed to, and that true/false partition the tenant exactly. // From f6d080be202d2858f02ceda0e0a93996ee101468 Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Mon, 5 Oct 2026 17:55:00 +0000 Subject: [PATCH 03/15] docs: spell out what the updated_since overlap covers (ENG-3420) The overlap also has to absorb the first response's latency and the clock of whatever stamped the Date header, and latest-fetch-wins only holds when syncs never overlap. Pin that the run-start watermark leaves unchanged records out, so the test fails if updated_since is ignored. --- .../core-concepts/filtering-and-sorting.mdx | 30 ++++++++++--------- tests/feedback_record_filters_test.go | 3 +- 2 files changed, 18 insertions(+), 15 deletions(-) diff --git a/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx b/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx index c44eb8c5..95568c8d 100644 --- a/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx +++ b/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx @@ -153,20 +153,22 @@ updated. Do not use the largest `updated_at` you received. A run that spans several pages reads each page at a different moment: a record you paged past early can change while later pages are still being read, and its new `updated_at` is then older than the newest one on your last page. A watermark -taken from the results skips that change. A watermark taken from the run's start does not, because -anything written after the run began is newer than it. - -- **Take the start time from Hub, not from your own clock.** Every response carries an HTTP `Date` - header; the one on the run's first response is the run's start on Hub's side, so drift between - your clock and Hub's cannot open a gap. -- **Keep an overlap of a few minutes, five is a safe default.** `updated_at` is when the writing - transaction *started*, so a write that was still committing when your run began carries a - slightly earlier timestamp. The overlap has to cover the longest write plus the difference between - Hub's and its database's clocks. -- **The latest fetch wins.** The overlap re-reads some records, so upsert by `id` and let the copy - from the most recent run replace the older one. Do not compare `updated_at` to pick a copy: two - writes to one record can commit in the opposite order to the one they started in, so a newer state - can carry an older timestamp. A later read always sees the later state. +taken from the results skips that change. A watermark taken from the run's start, minus the overlap +below, does not. + +- **Take the start time from the server, not from your own clock.** Every response carries an HTTP + `Date` header; use the one on the run's first response, so drift between your clock and the + server's cannot open a gap. +- **Keep an overlap of a few minutes, five is a safe default.** It has to cover three things: a + write that was still committing when the run began (`updated_at` is when the writing transaction + *started*), the time the first response took (`Date` is stamped after that page was read), and + the clock difference between whatever stamped `Date` (Hub, or a proxy in front of it) and the + database. +- **The latest fetch wins, so never run two syncs at once.** The overlap re-reads some records, so + upsert by `id` and let the copy from the most recent run replace the older one. Do not compare + `updated_at` to pick a copy: two writes to one record can commit in the opposite order to the one + they started in, so a newer state can carry an older timestamp. A later read always sees the later + state, which only holds if one run finishes before the next begins. ### Deleted records diff --git a/tests/feedback_record_filters_test.go b/tests/feedback_record_filters_test.go index 7443011c..b984f131 100644 --- a/tests/feedback_record_filters_test.go +++ b/tests/feedback_record_filters_test.go @@ -424,7 +424,7 @@ func TestFeedbackRecordFilters_UpdatedSinceWatermarkFromRunStart(t *testing.T) { ) first := env.seed(t) - env.seed(t) + middle := env.seed(t) // never changes, so the second run must leave it out last := env.seed(t) touch := func(id uuid.UUID) { @@ -502,6 +502,7 @@ func TestFeedbackRecordFilters_UpdatedSinceWatermarkFromRunStart(t *testing.T) { } assert.Contains(t, ids, first.ID, "a run-start watermark must return the change made behind the cursor") + assert.NotContains(t, ids, middle.ID, "a record unchanged since the run started must not be re-read") } // TestFeedbackRecordFilters_PresencePartitions verifies each presence filter reads the column it From b9167531878eb972daf22d432ec84c9f155a7316 Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Mon, 5 Oct 2026 17:57:55 +0000 Subject: [PATCH 04/15] docs: guard the deletion sweep in the updated_since parameter description (ENG-3420) --- openapi.yaml | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/openapi.yaml b/openapi.yaml index 6e7e9be9..aaed12d1 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -2733,8 +2733,10 @@ components: parameter for incremental extraction. For the next run, use the time the previous run started (its first response's Date header) minus a few minutes of overlap, not the largest updated_at received. Permanently deleted records are never returned: reconcile - deletions separately by listing record ids. updated_at is not a sort key; keep sorting - by collected_at or created_at, which stays stable while matching records change. + deletions separately by listing every record id, and remove only local records that + were not returned and were created before that listing started (its first response's + Date header) minus the overlap. updated_at is not a sort key; keep sorting by + collected_at or created_at, which stays stable while matching records change. schema: type: string format: date-time From cada82d8d1bd48982eb2c6f715dc6a9f36f3fbbe Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Tue, 6 Oct 2026 11:48:25 +0000 Subject: [PATCH 05/15] docs: state the measured write cost of the updated_at index (ENG-3420) Indexing updated_at removes HOT updates from feedback_records entirely, since every writer sets it, so translation writes and metadata-only PATCHes now maintain all 28 indexes rather than one more. Record the measured numbers, why the trade is still worth it, and the build's WAL burst. --- .../027_feedback_records_updated_at_index.sql | 26 ++++++++++++------- 1 file changed, 17 insertions(+), 9 deletions(-) diff --git a/migrations/027_feedback_records_updated_at_index.sql b/migrations/027_feedback_records_updated_at_index.sql index 9d9e8f9b..cc559b19 100644 --- a/migrations/027_feedback_records_updated_at_index.sql +++ b/migrations/027_feedback_records_updated_at_index.sql @@ -6,8 +6,9 @@ -- updated_since exists for incremental extraction: "every record created or changed since my last -- run". That query is narrow by design — a sync that runs hourly matches a sliver of the tenant — -- and without an index the planner can only satisfy it by walking --- idx_feedback_records_tenant_collected_at_id and discarding rows, so every page of every sync --- reads the whole tenant. That is the degradation migration 021 indexed created_at against, and it +-- idx_feedback_records_tenant_collected_at_id and discarding rows, or by a sequential scan of the +-- whole table (what it chose on a local 2M-row table), so every page of every sync reads at least +-- the whole tenant. That is the degradation migration 021 indexed created_at against, and it -- gets worse exactly as a tenant grows large enough to need incremental extraction at all. -- -- (tenant_id, updated_at), with no id and no DESC: unlike created_at, updated_at is never a sort @@ -15,14 +16,21 @@ -- listing still orders by collected_at or created_at; a narrow match set is fetched here and -- sorted, and a wide one falls back to walking the ordering index, which the planner picks by cost. -- --- Write cost, accepted deliberately: updated_at changes on every PATCH and every enrichment write, --- so each of those updates now maintains this index too. Sentiment, emotions and PATCH writes --- already touch indexed columns and were not HOT updates anyway; a translation write touched none --- and could be HOT until now. One extra index entry per enrichment write is the price of an --- extraction query that does not scan the tenant. +-- Write cost, accepted deliberately, and larger than "one more index": every writer sets updated_at, +-- so once it is indexed NO update to this table can be a HOT update, and a non-HOT update writes a +-- new entry into every index on the table (28 with this one). Sentiment and emotions writes, and +-- PATCHes touching indexed columns, were already non-HOT. Translation writes and metadata-only +-- PATCHes were not, and now pay the full index maintenance. Measured on a local 2M-row table, +-- translation-shaped single-row updates went from ~5.7-7.1k TPS (83-91% HOT) to ~1.1-1.2k TPS +-- (0% HOT). Accepted because enrichment is bound by LLM latency at tens of writes per second, far +-- below that ceiling, whereas without the index every updated_since query is a sequential scan of +-- the whole table, all tenants included (57-105 ms at 2M rows and growing with total data, against +-- ~0.1 ms with it). Expect more WAL and index bloat on translation-heavy deployments. -- --- Runs without a transaction because of CONCURRENTLY, and is re-runnable: DROP-then-CREATE replaces --- the INVALID index an interrupted CREATE INDEX CONCURRENTLY leaves behind (see 021). +-- Runs without a transaction because of CONCURRENTLY, so writes continue during the build, and is +-- re-runnable: DROP-then-CREATE replaces the INVALID index an interrupted CREATE INDEX CONCURRENTLY +-- leaves behind (see 021). The build itself is a burst of WAL; locally it took ~10 s on 2M rows and +-- caused a checkpoint stall, so a large deployment should expect a short write-latency spike. DROP INDEX CONCURRENTLY IF EXISTS idx_feedback_records_tenant_updated_at; CREATE INDEX CONCURRENTLY idx_feedback_records_tenant_updated_at ON feedback_records (tenant_id, updated_at); From 78c8753cdf5c884d01e1cbed8286e83523bbd306 Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Tue, 6 Oct 2026 11:52:32 +0000 Subject: [PATCH 06/15] docs: correct the index count and scan wording in migration 027 (ENG-3420) 29 indexes counting the primary key, and partial indexes only take an entry when the new row matches their predicate. The whole-table scan is what the planner picks for a large tenant, not for every query. --- .../027_feedback_records_updated_at_index.sql | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/migrations/027_feedback_records_updated_at_index.sql b/migrations/027_feedback_records_updated_at_index.sql index cc559b19..e684886c 100644 --- a/migrations/027_feedback_records_updated_at_index.sql +++ b/migrations/027_feedback_records_updated_at_index.sql @@ -17,15 +17,16 @@ -- sorted, and a wide one falls back to walking the ordering index, which the planner picks by cost. -- -- Write cost, accepted deliberately, and larger than "one more index": every writer sets updated_at, --- so once it is indexed NO update to this table can be a HOT update, and a non-HOT update writes a --- new entry into every index on the table (28 with this one). Sentiment and emotions writes, and --- PATCHes touching indexed columns, were already non-HOT. Translation writes and metadata-only --- PATCHes were not, and now pay the full index maintenance. Measured on a local 2M-row table, +-- so once it is indexed NO update to this table can be a HOT update, and a non-HOT update inserts +-- into every index on the table (29 with this one, counting the primary key), except the partial +-- indexes whose predicate the new row does not match. Sentiment and emotions writes, and PATCHes +-- touching indexed columns, were already non-HOT. Translation writes and metadata-only PATCHes were +-- not: each went from about zero index inserts to twenty-odd. Measured on a local 2M-row table, -- translation-shaped single-row updates went from ~5.7-7.1k TPS (83-91% HOT) to ~1.1-1.2k TPS -- (0% HOT). Accepted because enrichment is bound by LLM latency at tens of writes per second, far --- below that ceiling, whereas without the index every updated_since query is a sequential scan of --- the whole table, all tenants included (57-105 ms at 2M rows and growing with total data, against --- ~0.1 ms with it). Expect more WAL and index bloat on translation-heavy deployments. +-- below that ceiling, whereas without the index an updated_since query for a large tenant is a +-- sequential scan of the whole table, all tenants included (57-105 ms at 2M rows and growing with +-- total data, against ~0.1 ms with it). Expect more WAL and index bloat on translation-heavy deployments. -- -- Runs without a transaction because of CONCURRENTLY, so writes continue during the build, and is -- re-runnable: DROP-then-CREATE replaces the INVALID index an interrupted CREATE INDEX CONCURRENTLY From 1afb2704eee5b654c284d0db8e2225ddc7299639 Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Tue, 6 Oct 2026 12:04:14 +0000 Subject: [PATCH 07/15] docs: tie the updated_since overlap to the longest write (ENG-3420) The overlap has to exceed the longest transaction that writes a record, and Hub does not bound one. Say so, and have the full sweep upsert what it returns so it also repairs a change a stalled write slipped past the watermark. --- .../core-concepts/filtering-and-sorting.mdx | 22 +++++++++++-------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx b/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx index 95568c8d..96fa3961 100644 --- a/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx +++ b/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx @@ -159,12 +159,15 @@ below, does not. - **Take the start time from the server, not from your own clock.** Every response carries an HTTP `Date` header; use the one on the run's first response, so drift between your clock and the server's cannot open a gap. -- **Keep an overlap of a few minutes, five is a safe default.** It has to cover three things: a - write that was still committing when the run began (`updated_at` is when the writing transaction - *started*), the time the first response took (`Date` is stamped after that page was read), and - the clock difference between whatever stamped `Date` (Hub, or a proxy in front of it) and the - database. -- **The latest fetch wins, so never run two syncs at once.** The overlap re-reads some records, so +- **Keep an overlap longer than any write to a record can take.** `updated_at` is when the + writing transaction *started*, so a write that commits after your run began can carry a + timestamp from before it; the overlap has to exceed the longest such transaction, plus the time + the first response took (`Date` is stamped after that page was read) and the clock difference + between whatever stamped `Date` (Hub, or a proxy in front of it) and the database. Hub's writes + are short single-record transactions, so five minutes is a comfortable default, but Hub does not + enforce a limit on how long one may take. A write stalled for longer than the overlap is missed + by the incremental sync, which is one reason to keep the full sweep below. +- **The latest fetch wins, so never run two syncs at once** (the full sweep below included). The overlap re-reads some records, so upsert by `id` and let the copy from the most recent run replace the older one. Do not compare `updated_at` to pick a copy: two writes to one record can commit in the opposite order to the one they started in, so a newer state can carry an older timestamp. A later read always sees the later @@ -185,9 +188,10 @@ that were not returned **and whose `created_at` is earlier than that start minus That condition matters: a record created after the sweep began may have been stored by your incremental sync but never reached by the sweep, and deleting it would lose it for good. -This catches every way a record can disappear, including deletes by user and dataset purges. It -reads the whole tenant, so run it less often than the incremental sync, for example daily against -an hourly sync. +This catches every way a record can disappear, including deletes by user and dataset purges. The +sweep returns every record in full, so upsert what it returns as well: that also repairs any change +the incremental sync missed, such as a write that took longer than the overlap. It reads the whole +tenant, so run it less often than the incremental sync, for example daily against an hourly sync. ## Enrichment filters From 8c93e3b59c339b68e0804642f23fea8b1ab623ab Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Thu, 8 Oct 2026 18:00:13 +0000 Subject: [PATCH 08/15] perf: plan filtered list and count per execution (ENG-3420) pgx's default mode keeps a named prepared statement per connection, and Postgres plans one generically from its sixth execution. A generic plan cannot see the tenant or the window, so a large tenant's updated_since sync went from <1 ms to ~15 ms per page from the sixth call on. Run list, cursor pages and count on the unnamed statement so every execution gets a custom plan, and pin that with a single-connection integration test. --- .../repository/feedback_records_repository.go | 19 ++++++- tests/feedback_record_filters_test.go | 56 +++++++++++++++++++ 2 files changed, 73 insertions(+), 2 deletions(-) diff --git a/internal/repository/feedback_records_repository.go b/internal/repository/feedback_records_repository.go index dded55d9..ed048577 100644 --- a/internal/repository/feedback_records_repository.go +++ b/internal/repository/feedback_records_repository.go @@ -652,6 +652,21 @@ func scanBackfillTargetIDs(rows pgx.Rows, name string) ([]uuid.UUID, error) { return ids, nil } +// filterQueryArgs runs a filtered list or count on the unnamed prepared statement instead of pgx's +// default named one, so Postgres plans every execution with its actual parameter values. +// +// A named statement is planned generically from its sixth execution on a connection whenever the +// generic plan's estimated cost looks no worse than the custom ones, and a generic plan cannot see +// which tenant or how narrow a time window it is serving. On a large tenant among many small ones +// that turns an updated_since sync (ENG-3420) from an index scan of the matching rows into a walk of +// the tenant's ordering index that discards nearly everything: under 1 ms for the first five calls, +// ~15 ms from the sixth, on a local 200k-row tenant. Every filter on these endpoints is skewed by +// tenant size the same way, so the mode covers list, cursor pages and count alike. The cost is a +// plan per execution, still in one round trip because the statement description stays cached. +func filterQueryArgs(args []any) []any { + return append([]any{pgx.QueryExecModeCacheDescribe}, args...) +} + // Count returns the number of feedback records matching the given filters. func (r *FeedbackRecordsRepository) Count( ctx context.Context, filters *models.ListFeedbackRecordsFilters, @@ -662,7 +677,7 @@ func (r *FeedbackRecordsRepository) Count( } var count int - if err := r.db.QueryRow(ctx, query, args...).Scan(&count); err != nil { + if err := r.db.QueryRow(ctx, query, filterQueryArgs(args)...).Scan(&count); err != nil { return 0, fmt.Errorf("query feedback records count: %w", err) } @@ -1196,7 +1211,7 @@ func listUserFeedbackTenants( func (r *FeedbackRecordsRepository) fetchFeedbackRecords( ctx context.Context, query string, args ...any, ) ([]models.FeedbackRecord, error) { - rows, err := r.db.Query(ctx, query, args...) + rows, err := r.db.Query(ctx, query, filterQueryArgs(args)...) if err != nil { return nil, fmt.Errorf("failed to list feedback records: %w", err) } diff --git a/tests/feedback_record_filters_test.go b/tests/feedback_record_filters_test.go index b984f131..70d5293a 100644 --- a/tests/feedback_record_filters_test.go +++ b/tests/feedback_record_filters_test.go @@ -5,6 +5,7 @@ import ( "time" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -505,6 +506,61 @@ func TestFeedbackRecordFilters_UpdatedSinceWatermarkFromRunStart(t *testing.T) { assert.NotContains(t, ids, middle.ID, "a record unchanged since the run started must not be re-read") } +// TestFeedbackRecordFilters_ListAndCountAvoidNamedStatements pins the plan-cache fix behind +// updated_since (ENG-3420). pgx's default mode keeps a named prepared statement per connection, and +// Postgres plans one generically from its sixth execution; a generic plan cannot see which tenant or +// how narrow a window it serves, so a large tenant's sync degraded from an index scan of the matching +// rows to a walk of the whole tenant. Timings are too noisy to assert on, so this asserts the +// mechanism instead: on a single connection, list, cursor page and count leave no named statement. +func TestFeedbackRecordFilters_ListAndCountAvoidNamedStatements(t *testing.T) { + env := newFilterTestEnv(t, "unnamed-statements") + first := env.seed(t) + env.seed(t) + + cfg, err := config.Load() + require.NoError(t, err) + + single, err := database.NewPostgresPool(t.Context(), cfg.Database.URL, + database.PoolOption(func(c *pgxpool.Config) { c.MaxConns = 1 })) + require.NoError(t, err) + t.Cleanup(single.Close) + + repo := repository.NewFeedbackRecordsRepository(single) + tenant := env.tenant + since := first.CreatedAt.Add(-time.Hour) + + // Well past the five executions after which a named statement may switch to a generic plan. + for range 7 { + filters := &models.ListFeedbackRecordsFilters{TenantID: &tenant, UpdatedSince: &since, Limit: 1} + + _, _, err = repo.List(t.Context(), filters) + require.NoError(t, err) + + _, _, err = repo.ListAfterCursor(t.Context(), filters, first.CollectedAt, first.ID) + require.NoError(t, err) + + _, err = repo.Count(t.Context(), filters) + require.NoError(t, err) + } + + // Simple protocol, so this probe does not become a named statement itself. + var named []string + + rows, err := single.Query(t.Context(), + `SELECT statement FROM pg_prepared_statements WHERE statement ILIKE '%FROM feedback_records%'`, + pgx.QueryExecModeSimpleProtocol) + require.NoError(t, err) + + for rows.Next() { + var statement string + require.NoError(t, rows.Scan(&statement)) + named = append(named, statement) + } + + require.NoError(t, rows.Err()) + assert.Empty(t, named, "filtered list/count queries must run on the unnamed statement") +} + // TestFeedbackRecordFilters_PresencePartitions verifies each presence filter reads the column it // is supposed to, and that true/false partition the tenant exactly. // From 6dde0737e18eec4228f3d1310d46b8bc46116ae1 Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Thu, 8 Oct 2026 18:01:25 +0000 Subject: [PATCH 09/15] test: pin every filter's public name to openapi.yaml (ENG-3420) The integration tests set struct fields directly, so a form-tag typo would pass them while the decoder ignored the real parameter. Resolve the list and count operations' documented parameters and require them to match the struct's form tags in both directions, with timestamps as date-time. --- .../feedback_records_filters_openapi_test.go | 115 ++++++++++++++++++ 1 file changed, 115 insertions(+) diff --git a/internal/models/feedback_records_filters_openapi_test.go b/internal/models/feedback_records_filters_openapi_test.go index b2f88960..4099ed0c 100644 --- a/internal/models/feedback_records_filters_openapi_test.go +++ b/internal/models/feedback_records_filters_openapi_test.go @@ -7,6 +7,7 @@ import ( "runtime" "strings" "testing" + "time" "gopkg.in/yaml.v3" ) @@ -149,3 +150,117 @@ func loadOpenAPIParameters(t *testing.T) openAPIParameters { return spec } + +// openAPIOperationParameters is the slice of openapi.yaml naming each list operation's query +// parameters, either inline or by $ref to a shared component. +type openAPIOperationParameters struct { + Paths map[string]struct { + Get struct { + Parameters []struct { + Ref string `yaml:"$ref"` + Name string `yaml:"name"` + Schema struct { + Format string `yaml:"format"` + } `yaml:"schema"` + } `yaml:"parameters"` + } `yaml:"get"` + } `yaml:"paths"` + Components struct { + Parameters map[string]struct { + Name string `yaml:"name"` + Schema struct { + Format string `yaml:"format"` + } `yaml:"schema"` + } `yaml:"parameters"` + } `yaml:"components"` +} + +// TestOpenAPIDocumentsEveryFilterByItsFormName pins the public parameter names to the struct's form +// tags, in both directions. The integration tests set struct fields directly, so a tag typo such as +// `form:"updated_untl"` would pass every other test while the decoder silently ignored the real +// parameter — for updated_since (ENG-3420), a sync that reads the whole tenant every run. +// +// List must document every field; count every field except the four pagination and ordering ones, +// which it does not accept. Every documented parameter must in turn decode into a field, and every +// timestamp field must be documented as date-time. +func TestOpenAPIDocumentsEveryFilterByItsFormName(t *testing.T) { + t.Parallel() + + _, filename, _, ok := runtime.Caller(0) + if !ok { + t.Fatal("cannot resolve this test file's path") + } + + // #nosec G304 -- the repository-local spec path is derived from this test file's location. + contents, err := os.ReadFile(filepath.Join(filepath.Dir(filename), "..", "..", "openapi.yaml")) + if err != nil { + t.Fatalf("read openapi.yaml: %v", err) + } + + var spec openAPIOperationParameters + if err := yaml.Unmarshal(contents, &spec); err != nil { + t.Fatalf("parse openapi.yaml: %v", err) + } + + documented := func(path string) map[string]string { + formats := map[string]string{} + + for _, parameter := range spec.Paths[path].Get.Parameters { + if parameter.Ref == "" { + formats[parameter.Name] = parameter.Schema.Format + + continue + } + + component, found := spec.Components.Parameters[strings.TrimPrefix(parameter.Ref, "#/components/parameters/")] + if !found { + t.Fatalf("%s references %s, which openapi.yaml does not define", path, parameter.Ref) + } + + formats[component.Name] = component.Schema.Format + } + + return formats + } + + listOnly := map[string]bool{"sort": true, "order": true, "limit": true, "cursor": true} + timeType := reflect.TypeFor[*time.Time]() + formNames := map[string]bool{} + + for _, operation := range []struct { + path string + formats map[string]string + skipNames map[string]bool + }{ + {"/v1/feedback-records", documented("/v1/feedback-records"), nil}, + {"/v1/feedback-records/count", documented("/v1/feedback-records/count"), listOnly}, + } { + if len(operation.formats) == 0 { + t.Fatalf("%s documents no parameters; this test would check nothing", operation.path) + } + + for field := range reflect.TypeFor[ListFeedbackRecordsFilters]().Fields() { + name, _, _ := strings.Cut(field.Tag.Get("form"), ",") + formNames[name] = true + + if operation.skipNames[name] { + continue + } + + format, isDocumented := operation.formats[name] + if !isDocumented { + t.Fatalf("%s does not document %q, the form name of %s", operation.path, name, field.Name) + } + + if field.Type == timeType && format != "date-time" { + t.Fatalf("%s documents %q with format %q; %s is a timestamp", operation.path, name, format, field.Name) + } + } + + for name := range operation.formats { + if !formNames[name] { + t.Fatalf("%s documents %q, which no filter field decodes; the spec names a parameter the API ignores", operation.path, name) + } + } + } +} From 920c940b23907322893a97490691486fc70a375e Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Thu, 8 Oct 2026 18:26:47 +0000 Subject: [PATCH 10/15] docs: measure the translation burst, sync the API text and the Airbyte guide (ENG-3420) The common translation write makes no LLM call, so a backfill is a burst paced by worker concurrency. A 20k-record backfill through hub-worker drained at the same rate with and without the updated_at index (~105/s at concurrency 5, ~570-600/s at 50); the cost is 114 vs 81 MB of WAL plus bloat. The updated_since API description now carries the unbounded-write caveat and the upserting sweep, and the Airbyte guide stops calling updated_at a perfect cursor. --- docs/src/content/docs/guides/hub-airbyte.mdx | 12 +++++++++++- .../027_feedback_records_updated_at_index.sql | 19 +++++++++++++------ openapi.yaml | 15 +++++++++------ 3 files changed, 33 insertions(+), 13 deletions(-) diff --git a/docs/src/content/docs/guides/hub-airbyte.mdx b/docs/src/content/docs/guides/hub-airbyte.mdx index 5bbcf402..fe1cc295 100644 --- a/docs/src/content/docs/guides/hub-airbyte.mdx +++ b/docs/src/content/docs/guides/hub-airbyte.mdx @@ -21,7 +21,7 @@ Hub stores all data in a single PostgreSQL database. **Airbyte** is an open-sour - **One source, many destinations** - Hub's data is a standard PostgreSQL table. Airbyte's Postgres source connector reads it, and any of Airbyte's [warehouse destinations](https://docs.airbyte.com/integrations/destinations) (Snowflake, Databricks, BigQuery, Redshift, ClickHouse, ...) can receive it. - **Open source and self-hostable** - [Airbyte OSS](https://docs.airbyte.com/) runs on your own infrastructure; no third-party SaaS is involved. -- **Incremental sync out of the box** - Hub's `updated_at` column is a perfect cursor for incremental replication; you only move new and changed rows after the initial backfill. +- **Incremental sync out of the box** - Hub's `updated_at` column lets incremental replication move only new and changed rows after the initial backfill. It is not a perfect cursor, though: see the caution in Step 5. - **Schema-aware** - Airbyte maps Hub's columns to native warehouse types (including `metadata` JSONB to Snowflake `VARIANT`, BigQuery `JSON`, Redshift `SUPER`, etc.). + + ## Query in the destination After the first sync completes, you can query `feedback_records` in the destination warehouse using the same dimensions and metrics as everywhere else in Hub. The `metadata` JSONB column is preserved as a native semi-structured type in the destination: diff --git a/migrations/027_feedback_records_updated_at_index.sql b/migrations/027_feedback_records_updated_at_index.sql index e684886c..08c599b4 100644 --- a/migrations/027_feedback_records_updated_at_index.sql +++ b/migrations/027_feedback_records_updated_at_index.sql @@ -21,12 +21,19 @@ -- into every index on the table (29 with this one, counting the primary key), except the partial -- indexes whose predicate the new row does not match. Sentiment and emotions writes, and PATCHes -- touching indexed columns, were already non-HOT. Translation writes and metadata-only PATCHes were --- not: each went from about zero index inserts to twenty-odd. Measured on a local 2M-row table, --- translation-shaped single-row updates went from ~5.7-7.1k TPS (83-91% HOT) to ~1.1-1.2k TPS --- (0% HOT). Accepted because enrichment is bound by LLM latency at tens of writes per second, far --- below that ceiling, whereas without the index an updated_since query for a large tenant is a --- sequential scan of the whole table, all tenants included (57-105 ms at 2M rows and growing with --- total data, against ~0.1 ms with it). Expect more WAL and index bloat on translation-heavy deployments. +-- not: each went from about zero index inserts to twenty-odd. As a raw ceiling, translation-shaped +-- single-row updates on a local 2M-row table went from ~5.7-7.1k TPS (83-91% HOT) to ~1.1-1.2k TPS +-- (0% HOT). +-- +-- That ceiling is not what the pipeline hits. The common translation write makes no LLM call at +-- all (an unset source language copies value_text), so a bulk import or a target-language backfill +-- is a burst of these writes paced by worker concurrency. Measured end to end through hub-worker on +-- a 20k-record backfill, the drain rate was the same with and without this index: ~105/s at the +-- default TRANSLATION_MAX_CONCURRENT of 5, ~570-600/s at 50, because per-job overhead dominates the +-- write. The cost shows up as storage traffic instead: 114 MB of WAL against 81 MB for those 20k +-- copies (job bookkeeping included), and index bloat for autovacuum to clean up. Accepted, because +-- without the index an updated_since query for a large tenant is a sequential scan of the whole +-- table, all tenants included (57-105 ms at 2M rows and growing with total data, ~0.1 ms with it). -- -- Runs without a transaction because of CONCURRENTLY, so writes continue during the build, and is -- re-runnable: DROP-then-CREATE replaces the INVALID index an interrupted CREATE INDEX CONCURRENTLY diff --git a/openapi.yaml b/openapi.yaml index aaed12d1..6f6e99bc 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -2731,12 +2731,15 @@ components: every write to a record — create, update, and every enrichment result (translation, sentiment, emotions) — so this selects records created OR changed since a point: the parameter for incremental extraction. For the next run, use the time the previous run - started (its first response's Date header) minus a few minutes of overlap, not the - largest updated_at received. Permanently deleted records are never returned: reconcile - deletions separately by listing every record id, and remove only local records that - were not returned and were created before that listing started (its first response's - Date header) minus the overlap. updated_at is not a sort key; keep sorting by - collected_at or created_at, which stays stable while matching records change. + started (its first response's Date header) minus an overlap, not the largest updated_at + received. The overlap must outlast the longest write to a record; Hub does not bound + that, so a few minutes is a default rather than a guarantee. Permanently deleted records + are never returned: periodically list every record, upsert what it returns (which also + repairs any change a slow write slipped past the overlap), and remove only local records + that were not returned and were created before that listing started (its first + response's Date header) minus the overlap. updated_at is not a sort key; keep sorting by + collected_at or created_at, which stays stable while matching records change. See + "Incremental extraction" in the filtering and sorting guide. schema: type: string format: date-time From 7c18730f4b1561a9d2cf1c79955f919d03bc7317 Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Thu, 8 Oct 2026 18:26:55 +0000 Subject: [PATCH 11/15] test: walk the updated_since traversal in both orders (ENG-3420) The comment named a skip that order=asc cannot produce. Run the traversal under asc and desc, touch the first-returned and last-unreached record in each, and say precisely what an updated_at sort key would do to each. --- tests/feedback_record_filters_test.go | 80 ++++++++++++++++----------- 1 file changed, 47 insertions(+), 33 deletions(-) diff --git a/tests/feedback_record_filters_test.go b/tests/feedback_record_filters_test.go index 70d5293a..50e3533e 100644 --- a/tests/feedback_record_filters_test.go +++ b/tests/feedback_record_filters_test.go @@ -1,6 +1,7 @@ package tests import ( + "slices" "testing" "time" @@ -369,45 +370,58 @@ func TestFeedbackRecordFilters_UpdatedSincePaginationSurvivesUpdates(t *testing. require.NoError(t, err) } - seen := make(map[uuid.UUID]int, records) - nextCursor := "" + // Seeded in created_at order, so this is the traversal order for asc; desc walks it backwards. + for _, order := range []models.SortOrder{models.SortOrderAsc, models.SortOrderDesc} { + t.Run(string(order), func(t *testing.T) { + inOrder := slices.Clone(ordered) + if order == models.SortOrderDesc { + slices.Reverse(inOrder) + } - for page := range records { - tenant := env.tenant - resp, err := svc.ListFeedbackRecords(t.Context(), &models.ListFeedbackRecordsFilters{ - TenantID: &tenant, - UpdatedSince: &cutoff, - Sort: models.SortFieldCreatedAt, - Order: models.SortOrderAsc, - Limit: pageLimit, - Cursor: nextCursor, - }) - require.NoError(t, err, "page %d", page) + seen := make(map[uuid.UUID]int, records) + nextCursor := "" - for _, record := range resp.Data { - seen[record.ID]++ - } + for page := range records { + tenant := env.tenant + resp, err := svc.ListFeedbackRecords(t.Context(), &models.ListFeedbackRecordsFilters{ + TenantID: &tenant, + UpdatedSince: &cutoff, + Sort: models.SortFieldCreatedAt, + Order: order, + Limit: pageLimit, + Cursor: nextCursor, + }) + require.NoError(t, err, "page %d", page) - if page == 0 { - // Change one record already returned and one not reached yet. Under sort=updated_at - // the second would jump behind the cursor and be skipped. - touch(ordered[0]) - touch(ordered[records-1]) - } + for _, record := range resp.Data { + seen[record.ID]++ + } - nextCursor = resp.NextCursor - if nextCursor == "" { - break - } - } + if page == 0 { + // Change the first record returned and the last one not reached yet. Their + // created_at, the sort key, does not move, so each is still returned once. Were + // updated_at the sort key, both would move to the newest end of the ordering: under + // asc the first would come round again (returned twice), and under desc the last + // would land behind the cursor (never returned). + touch(inOrder[0]) + touch(inOrder[records-1]) + } - require.Empty(t, nextCursor, "pagination did not terminate") - assert.NotContains(t, seen, before.ID, "a record last written before updated_since must not be returned") - assert.Len(t, seen, records, "every record must be returned") + nextCursor = resp.NextCursor + if nextCursor == "" { + break + } + } + + require.Empty(t, nextCursor, "pagination did not terminate") + assert.NotContains(t, seen, before.ID, "a record last written before updated_since must not be returned") + assert.Len(t, seen, records, "every record must be returned") - for id, times := range seen { - assert.True(t, want[id], "record %s does not belong to this traversal", id) - assert.Equal(t, 1, times, "record %s returned %d times", id, times) + for id, times := range seen { + assert.True(t, want[id], "record %s does not belong to this traversal", id) + assert.Equal(t, 1, times, "record %s returned %d times", id, times) + } + }) } } From 2cbe6ebcf08799b27934b0b61f91a59745af7296 Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Fri, 9 Oct 2026 09:03:23 +0000 Subject: [PATCH 12/15] feat: sort feedback records by updated_at, ascending only (ENG-3420) Filtering on updated_at while ordering by created_at left the planner guessing the match count. It multiplies the tenant's share by the window's share, cannot see that recent activity belongs to other tenants, and walked a 200k-record tenant per page (29-61 ms) even with per-call plans. Allow sort=updated_at with order=asc only: a record that changes mid-run moves ahead of the cursor and comes back (a duplicate to upsert), never behind it. Descending, or the default order, is a 400 in validation and in the repository. Index 027 becomes (tenant_id, updated_at, id), and the ascending keyset uses a row comparison so it is an index range: every page of a sync is O(limit) at any depth (0.04-0.2 ms vs 8.8 ms 150k rows deep). Docs and the spec now recommend sort=updated_at&order=asc for syncs. --- .../core-concepts/filtering-and-sorting.mdx | 35 +++--- internal/api/validation/validation.go | 15 ++- internal/api/validation/validation_test.go | 34 ++++++ internal/models/feedback_records.go | 2 + internal/models/feedback_records_filters.go | 32 ++++-- .../models/feedback_records_filters_test.go | 4 +- internal/repository/list_ordering.go | 37 ++++-- internal/repository/list_ordering_test.go | 18 ++- .../service/feedback_records_service_test.go | 3 + .../027_feedback_records_updated_at_index.sql | 23 ++-- openapi.yaml | 31 ++--- tests/feedback_record_filters_test.go | 106 ++++++++++-------- 12 files changed, 236 insertions(+), 104 deletions(-) diff --git a/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx b/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx index 96fa3961..09b8f080 100644 --- a/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx +++ b/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx @@ -132,29 +132,35 @@ new and changed records: Every writer takes the timestamp from the database clock, at the moment its transaction starts. ```bash -# everything created or changed since the last run, oldest first +# everything created or changed since the last run, in the order it changed curl -G "$HUB/v1/feedback-records" \ -H "Authorization: Bearer $HUB_API_KEY" \ -d tenant_id=org-123 \ -d updated_since=2026-10-05T07:55:00Z \ - -d sort=created_at \ + -d sort=updated_at \ -d order=asc \ -d limit=1000 ``` -**Sort by `created_at` or `collected_at`, not by `updated_at`.** `updated_at` is not a sort option, -on purpose: it changes under you, so a record could jump past the cursor and be skipped. The sort -columns never change, so every page stays stable even while records in the result are being -updated. +**Sort by `updated_at`, ascending.** Every page is then one range of the `updated_at` index, so a +sync reads only the records it returns, however large the tenant. Filtering on `updated_since` while +sorting by `created_at` also works, but the database cannot tell how few records changed and may +read the whole tenant for every page. + +Ascending is the only direction `updated_at` can be sorted in, because `updated_at` changes while you +page. A record that changes mid-run gets a newer `updated_at`, which moves it *ahead* of the cursor: +it comes back later in the same run, and you upsert it again. Descending would move it *behind* the +cursor, where it would be skipped, so `sort=updated_at` with `order=desc` — or with no `order`, which +defaults to `desc` — is a `400`. ### Choosing the next `updated_since` **Note when the run starts, and use that time minus an overlap as the next run's `updated_since`.** -Do not use the largest `updated_at` you received. A run that spans several pages reads each page at -a different moment: a record you paged past early can change while later pages are still being -read, and its new `updated_at` is then older than the newest one on your last page. A watermark -taken from the results skips that change. A watermark taken from the run's start, minus the overlap -below, does not. +Do not use the largest `updated_at` you received. `updated_at` is when a write *started*, and a +write becomes visible only when it commits: one that was still in flight when your run read past its +timestamp is missing from this run, and its `updated_at` is older than the newest one you received. +A watermark taken from the results skips it for good. A watermark taken from the run's start, minus +the overlap below, reads it next time. - **Take the start time from the server, not from your own clock.** Every response carries an HTTP `Date` header; use the one on the run's first response, so drift between your clock and the @@ -218,9 +224,10 @@ record tagged `{joy, anger}` both match `?emotions=joy&emotions=anger`. ## Sorting -`sort` accepts `collected_at` (the default) or `created_at`; `order` accepts `asc` or `desc` -(default `desc`). Records tied on the sort column are ordered by `id` ascending, so the sequence is -always total and stable. +`sort` accepts `collected_at` (the default), `created_at` or `updated_at`; `order` accepts `asc` or +`desc` (default `desc`). Records tied on the sort column are ordered by `id` ascending, so the sequence +is always total and stable. `updated_at` is accepted only with `order=asc` — see +[Incremental extraction](#incremental-extraction). Sort by the column you filter on. Filtering `created_at` while ordering by `collected_at` cannot use the index for the ordering, and gets slower as a dataset grows. diff --git a/internal/api/validation/validation.go b/internal/api/validation/validation.go index 2bfd8ca1..7e939bbc 100644 --- a/internal/api/validation/validation.go +++ b/internal/api/validation/validation.go @@ -160,7 +160,13 @@ func init() { // parameter. const rangeBoundsTag = "range_bounds" -// validateListFeedbackRecordsFilters rejects inverted min/max filter pairs. +// updatedAtAscendingTag marks sort=updated_at combined with an order other than asc. Struct-level for +// the same reason as rangeBoundsTag: the rule is a relationship between two parameters, and it is +// attributed to order, the one the client must change. +const updatedAtAscendingTag = "updated_at_ascending" + +// validateListFeedbackRecordsFilters rejects inverted min/max filter pairs, and sort=updated_at with +// any order but asc. // // An inverted range matches zero rows, so without this a swapped pair returns an empty page that // the caller reads as "there is no such feedback" rather than "your filter is backwards". @@ -175,6 +181,11 @@ func validateListFeedbackRecordsFilters(structLevel validator.StructLevel) { // the client sent, not the Go identifier. structLevel.ReportError(inverted.MinValue, inverted.MinParam, inverted.StructField, rangeBoundsTag, inverted.MaxParam) } + + // An omitted order defaults to desc, so sort=updated_at needs order=asc spelled out. + if filters.Sort == models.SortFieldUpdatedAt && filters.Order != models.SortOrderAsc { + structLevel.ReportError(filters.Order, "order", "Order", updatedAtAscendingTag, "") + } } // registerEnumSliceTypes registers decoders for the repeatable enum query filters. @@ -382,6 +393,8 @@ func FormatFieldError(fieldErr validator.FieldError) string { case rangeBoundsTag: // Param carries the upper bound's parameter name, set by the struct-level validator. return "must be less than or equal to " + fieldErr.Param() + case updatedAtAscendingTag: + return models.UpdatedAtAscendingOnlyReason case "uuid": return "must be a valid UUID" case "rfc3339": diff --git a/internal/api/validation/validation_test.go b/internal/api/validation/validation_test.go index a1d9bb48..489be461 100644 --- a/internal/api/validation/validation_test.go +++ b/internal/api/validation/validation_test.go @@ -194,6 +194,40 @@ func TestDecodeEnumSliceQueryParams(t *testing.T) { // TestValidateInvertedRangeFilters verifies an inverted min/max pair is a 400 naming the lower // bound, rather than an empty result page the caller reads as "there is no such feedback". +// TestValidateUpdatedAtSortIsAscendingOnly covers the struct-level rule that sort=updated_at needs +// order=asc spelled out: descending would skip records that change mid-traversal, and an omitted +// order defaults to desc. The 400 names order, the parameter the client has to change. +func TestValidateUpdatedAtSortIsAscendingOnly(t *testing.T) { + tenant := "org-123" + + for _, order := range []models.SortOrder{"", models.SortOrderDesc} { + t.Run("rejects order="+string(order), func(t *testing.T) { + err := ValidateStruct(models.ListFeedbackRecordsFilters{ + TenantID: &tenant, Sort: models.SortFieldUpdatedAt, Order: order, + }) + require.Error(t, err) + + var validationErrs validator.ValidationErrors + require.ErrorAs(t, err, &validationErrs) + require.Len(t, validationErrs, 1) + assert.Equal(t, "order", FieldPath(validationErrs[0])) + assert.Equal(t, models.UpdatedAtAscendingOnlyReason, FormatFieldError(validationErrs[0])) + }) + } + + t.Run("accepts order=asc", func(t *testing.T) { + require.NoError(t, ValidateStruct(models.ListFeedbackRecordsFilters{ + TenantID: &tenant, Sort: models.SortFieldUpdatedAt, Order: models.SortOrderAsc, + })) + }) + + t.Run("leaves the other sorts' default order alone", func(t *testing.T) { + require.NoError(t, ValidateStruct(models.ListFeedbackRecordsFilters{ + TenantID: &tenant, Sort: models.SortFieldCreatedAt, + })) + }) +} + func TestValidateInvertedRangeFilters(t *testing.T) { tenant := "org-123" diff --git a/internal/models/feedback_records.go b/internal/models/feedback_records.go index c1e90a26..5b581903 100644 --- a/internal/models/feedback_records.go +++ b/internal/models/feedback_records.go @@ -473,6 +473,8 @@ func (r *FeedbackRecord) SortValue(field SortField) time.Time { return r.CreatedAt case SortFieldCollectedAt: return r.CollectedAt + case SortFieldUpdatedAt: + return r.UpdatedAt default: return r.CollectedAt } diff --git a/internal/models/feedback_records_filters.go b/internal/models/feedback_records_filters.go index 7e276666..49f9fa9f 100644 --- a/internal/models/feedback_records_filters.go +++ b/internal/models/feedback_records_filters.go @@ -9,15 +9,20 @@ import ( // SortField is the column a feedback record listing is ordered by (ENG-2059). // -// Every member must name a column that is both NOT NULL and IMMUTABLE after insert: -// - NOT NULL, because the keyset predicate has no NULL handling — `col < $t` is NULL for a NULL -// row, which would drop it from every page after the first. -// - immutable, because a mutable sort key lets a row move across the cursor between pages and be -// silently skipped. This is why updated_at is deliberately absent: the enrichment workers -// (SetTranslation, SetSentiment, writeEmotions) and every PATCH bump it, so `sort=updated_at` -// would lose rows under ordinary traffic. Incremental extraction FILTERS on it instead -// (updated_since, ENG-3420) while ordering by an immutable column, which keeps every page -// stable however often the matching rows change. +// Every member must name a NOT NULL column, because the keyset predicate has no NULL handling — +// `col < $t` is NULL for a NULL row, which would drop it from every page after the first. +// +// Every member must also be immutable after insert, with one deliberate exception. A mutable sort +// key lets a row move across the cursor between pages. collected_at and created_at never change. +// updated_at does — every PATCH and enrichment write bumps it — and is allowed in ONE direction +// only (ENG-3420): ascending, a changed row moves AHEAD of the cursor, so it is returned again +// (a duplicate the caller upserts away) but never skipped. Descending it would move behind the +// cursor and be lost, which is why resolveListOrdering and the HTTP validator both reject +// sort=updated_at with any order but asc. It exists because filtering on updated_at while ordering +// by another column leaves the planner guessing how many rows match — it multiplies the tenant's +// share by the window's share and cannot see that recent activity belongs to other tenants — and +// a wrong guess walks the whole tenant per page. Ordered by (updated_at, id) on the +// (tenant_id, updated_at, id) index, every page of an incremental sync is O(limit). // // Adding a member requires a case in the repository's resolveListOrdering (token -> SQL column) // and in FeedbackRecord.SortValue (token -> record field). Both switches are exhaustive-linted, so @@ -28,8 +33,15 @@ type SortField string const ( SortFieldCollectedAt SortField = "collected_at" SortFieldCreatedAt SortField = "created_at" + // SortFieldUpdatedAt is ascending-only; see SortField. + SortFieldUpdatedAt SortField = "updated_at" ) +// UpdatedAtAscendingOnlyReason is the client-facing reason for rejecting sort=updated_at with an order +// other than asc (the default order is desc, so the order must be given explicitly). +const UpdatedAtAscendingOnlyReason = "must be asc when sort=updated_at: a record that changes while you page would " + + "otherwise move behind the cursor and be skipped" + // SortOrder is the direction of a feedback record listing. type SortOrder string @@ -134,7 +146,7 @@ type ListFeedbackRecordsFilters struct { SentimentScoreMin *float64 `form:"sentiment_score_min" validate:"omitempty,gte=-1,lte=1"` SentimentScoreMax *float64 `form:"sentiment_score_max" validate:"omitempty,gte=-1,lte=1"` - Sort SortField `form:"sort" validate:"omitempty,oneof=collected_at created_at"` + Sort SortField `form:"sort" validate:"omitempty,oneof=collected_at created_at updated_at"` Order SortOrder `form:"order" validate:"omitempty,oneof=asc desc"` Limit int `form:"limit" validate:"omitempty,min=1,max=1000"` diff --git a/internal/models/feedback_records_filters_test.go b/internal/models/feedback_records_filters_test.go index 17316289..d6f7dcfc 100644 --- a/internal/models/feedback_records_filters_test.go +++ b/internal/models/feedback_records_filters_test.go @@ -162,7 +162,8 @@ func TestValidEnumValuesStrings(t *testing.T) { func TestFeedbackRecord_SortValue(t *testing.T) { collected := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) created := time.Date(2026, 6, 1, 0, 0, 0, 0, time.UTC) - record := &FeedbackRecord{CollectedAt: collected, CreatedAt: created} + updated := time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC) + record := &FeedbackRecord{CollectedAt: collected, CreatedAt: created, UpdatedAt: updated} tests := []struct { name string @@ -171,6 +172,7 @@ func TestFeedbackRecord_SortValue(t *testing.T) { }{ {name: "collected_at", field: SortFieldCollectedAt, want: collected}, {name: "created_at", field: SortFieldCreatedAt, want: created}, + {name: "updated_at", field: SortFieldUpdatedAt, want: updated}, {name: "unknown falls back to the default sort", field: SortField("nonsense"), want: collected}, } diff --git a/internal/repository/list_ordering.go b/internal/repository/list_ordering.go index 9519e502..97052b21 100644 --- a/internal/repository/list_ordering.go +++ b/internal/repository/list_ordering.go @@ -7,6 +7,10 @@ import ( "github.com/formbricks/hub/internal/models" ) +// updatedAtAscendingOnlyMessage is the reason reported when sort=updated_at comes with any order but +// asc, including the default (desc). Shared with the HTTP validator so both layers say the same. +const updatedAtAscendingOnlyMessage = models.UpdatedAtAscendingOnlyReason + // listOrdering is the resolved ORDER BY for a feedback-record listing. Only resolveListOrdering // produces one, so no identifier it carries can originate from a request value. type listOrdering struct { @@ -43,9 +47,19 @@ func resolveListOrdering(filters *models.ListFeedbackRecordsFilters) (listOrderi column = colCollectedAt case models.SortFieldCreatedAt: column = colCreatedAt + case models.SortFieldUpdatedAt: + // Ascending only, and checked here as well as in HTTP validation: the repository is also + // reached from Go callers, and a descending updated_at traversal skips rows that change + // mid-traversal (see models.SortField). + if order != models.SortOrderAsc { + return listOrdering{}, huberrors.NewValidationError("order", updatedAtAscendingOnlyMessage) + } + + column = colUpdatedAt default: return listOrdering{}, huberrors.NewValidationError("sort", - "must be one of: "+string(models.SortFieldCollectedAt)+", "+string(models.SortFieldCreatedAt)) + "must be one of: "+string(models.SortFieldCollectedAt)+", "+string(models.SortFieldCreatedAt)+ + ", "+string(models.SortFieldUpdatedAt)) } switch order { @@ -66,7 +80,7 @@ func resolveListOrdering(filters *models.ListFeedbackRecordsFilters) (listOrderi // comparison keeps one meaning. Ties are common here: collected_at is client-supplied, so every // answer of one submission typically carries the same value. // -// No NULLS FIRST/LAST: both sortable columns are NOT NULL, and an explicit clause disagreeing with +// No NULLS FIRST/LAST: every sortable column is NOT NULL, and an explicit clause disagreeing with // how the index was declared would make the index unusable for the ordering. func (o listOrdering) orderByClause() string { if o.desc { @@ -79,14 +93,21 @@ func (o listOrdering) orderByClause() string { // keysetPredicate renders "strictly after the cursor row" for this ordering, with $timeArg bound // to the cursor's sort-column value and $idArg to its ID. // -// DESC: the next row is further down the column, or tied on it with a larger id. ASC is the mirror -// image. The tiebreak is `id >` in both, because the tiebreak direction is `id ASC` in both. +// ASC: (col, id) > ($t, $id), a row comparison. Both parts ascend, so it means exactly "further up +// the column, or tied on it with a larger id" — and unlike the equivalent OR it is an index range +// condition, so a page deep into the traversal starts at the cursor instead of reading and +// discarding every row before it. On the (tenant_id, updated_at, id) index that keeps every page of +// a sort=updated_at sync O(limit) however deep it is (ENG-3420): a page 150k rows in read 5 buffers +// instead of discarding 150k rows. +// +// DESC: the next row is further down the column, or tied on it with a larger id. The tiebreak +// still ascends (`id ASC` in every ORDER BY), so the two parts run in opposite directions and no +// row comparison can express it; the OR form is the only correct one. func (o listOrdering) keysetPredicate(timeArg, idArg int) string { - comparison := "<" if !o.desc { - comparison = ">" + return fmt.Sprintf("(%s, id) > ($%d, $%d)", o.column, timeArg, idArg) } - return fmt.Sprintf("(%s %s $%d OR (%s = $%d AND id > $%d))", - o.column, comparison, timeArg, o.column, timeArg, idArg) + return fmt.Sprintf("(%s < $%d OR (%s = $%d AND id > $%d))", + o.column, timeArg, o.column, timeArg, idArg) } diff --git a/internal/repository/list_ordering_test.go b/internal/repository/list_ordering_test.go index 9ff8c466..aa96e3fa 100644 --- a/internal/repository/list_ordering_test.go +++ b/internal/repository/list_ordering_test.go @@ -28,6 +28,10 @@ func TestResolveListOrdering(t *testing.T) { name: "created_at asc", sort: models.SortFieldCreatedAt, order: models.SortOrderAsc, wantColumn: colCreatedAt, wantDesc: false, }, + { + name: "updated_at asc", sort: models.SortFieldUpdatedAt, order: models.SortOrderAsc, + wantColumn: colUpdatedAt, wantDesc: false, + }, } for _, tt := range tests { @@ -63,6 +67,10 @@ func TestResolveListOrdering_RejectsUnknownTokens(t *testing.T) { }{ {name: "unknown sort", sort: models.SortField("value_number"), order: models.SortOrderDesc}, {name: "unknown order", sort: models.SortFieldCreatedAt, order: models.SortOrder("sideways")}, + // updated_at is ascending-only (models.SortField): descending, a record that changes + // mid-traversal moves behind the cursor and is skipped. An omitted order defaults to desc. + {name: "updated_at desc", sort: models.SortFieldUpdatedAt, order: models.SortOrderDesc}, + {name: "updated_at with the default order", sort: models.SortFieldUpdatedAt}, { name: "injection attempt", sort: models.SortField(`collected_at"; DROP TABLE feedback_records--`), @@ -122,7 +130,7 @@ func TestOrderingRendering(t *testing.T) { name: "collected_at asc", ordering: listOrdering{column: colCollectedAt, desc: false}, wantOrderBy: " ORDER BY collected_at ASC, id ASC", - wantPredicate: "(collected_at > $1 OR (collected_at = $1 AND id > $2))", + wantPredicate: "(collected_at, id) > ($1, $2)", }, { name: "created_at desc", @@ -134,7 +142,13 @@ func TestOrderingRendering(t *testing.T) { name: "created_at asc", ordering: listOrdering{column: colCreatedAt, desc: false}, wantOrderBy: " ORDER BY created_at ASC, id ASC", - wantPredicate: "(created_at > $1 OR (created_at = $1 AND id > $2))", + wantPredicate: "(created_at, id) > ($1, $2)", + }, + { + name: "updated_at asc", + ordering: listOrdering{column: colUpdatedAt, desc: false}, + wantOrderBy: " ORDER BY updated_at ASC, id ASC", + wantPredicate: "(updated_at, id) > ($1, $2)", }, } diff --git a/internal/service/feedback_records_service_test.go b/internal/service/feedback_records_service_test.go index 86993d47..eb4d6455 100644 --- a/internal/service/feedback_records_service_test.go +++ b/internal/service/feedback_records_service_test.go @@ -1016,11 +1016,13 @@ func listPage() []models.FeedbackRecord { ID: uuid.MustParse("018e1234-0000-7000-8000-000000000001"), CollectedAt: time.Date(2026, 1, 2, 0, 0, 0, 0, time.UTC), CreatedAt: time.Date(2026, 6, 2, 0, 0, 0, 0, time.UTC), + UpdatedAt: time.Date(2026, 8, 2, 0, 0, 0, 0, time.UTC), } last := models.FeedbackRecord{ ID: uuid.MustParse("018e1234-0000-7000-8000-000000000002"), CollectedAt: time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC), CreatedAt: time.Date(2026, 6, 1, 0, 0, 0, 0, time.UTC), + UpdatedAt: time.Date(2026, 8, 1, 0, 0, 0, 0, time.UTC), } return []models.FeedbackRecord{first, last} @@ -1062,6 +1064,7 @@ func TestListFeedbackRecords_NextCursorBoundsTheSortColumn(t *testing.T) { }{ {name: "collected_at", sort: models.SortFieldCollectedAt, order: models.SortOrderDesc, want: last.CollectedAt}, {name: "created_at", sort: models.SortFieldCreatedAt, order: models.SortOrderAsc, want: last.CreatedAt}, + {name: "updated_at", sort: models.SortFieldUpdatedAt, order: models.SortOrderAsc, want: last.UpdatedAt}, } for _, tt := range tests { diff --git a/migrations/027_feedback_records_updated_at_index.sql b/migrations/027_feedback_records_updated_at_index.sql index 08c599b4..09c873dd 100644 --- a/migrations/027_feedback_records_updated_at_index.sql +++ b/migrations/027_feedback_records_updated_at_index.sql @@ -5,16 +5,19 @@ -- -- updated_since exists for incremental extraction: "every record created or changed since my last -- run". That query is narrow by design — a sync that runs hourly matches a sliver of the tenant — --- and without an index the planner can only satisfy it by walking --- idx_feedback_records_tenant_collected_at_id and discarding rows, or by a sequential scan of the --- whole table (what it chose on a local 2M-row table), so every page of every sync reads at least --- the whole tenant. That is the degradation migration 021 indexed created_at against, and it --- gets worse exactly as a tenant grows large enough to need incremental extraction at all. +-- and without this index it can only be answered by walking the tenant's collected_at/created_at +-- ordering index and discarding rows, or by a sequential scan of the whole table, so every page of +-- every sync reads at least the whole tenant. -- --- (tenant_id, updated_at), with no id and no DESC: unlike created_at, updated_at is never a sort --- key (it is mutable, see models.SortField), so this index only has to find the matching rows. The --- listing still orders by collected_at or created_at; a narrow match set is fetched here and --- sorted, and a wide one falls back to walking the ordering index, which the planner picks by cost. +-- (tenant_id, updated_at, id) serves the sync as one ordered index range: sort=updated_at&order=asc +-- (the one direction updated_at may be sorted in, see models.SortField) with the keyset tiebreak on +-- id, so every page reads LIMIT rows and stops. Filtering on updated_at while ordering by created_at +-- would be served by the same index only when the planner guesses the match count right, and it +-- guesses wrong exactly in the case that matters: it multiplies the tenant's share of the table by +-- the window's share and cannot see that the recent activity belongs to other tenants. Measured on +-- a 200k-record tenant among 2,000 small ones with 69 recent changes, it estimated ~3,000 matches, +-- walked the ordering index and read all 200k rows per page (29-61 ms), even with a per-execution +-- plan. The same request ordered by updated_at reads the 69. -- -- Write cost, accepted deliberately, and larger than "one more index": every writer sets updated_at, -- so once it is indexed NO update to this table can be a HOT update, and a non-HOT update inserts @@ -41,7 +44,7 @@ -- caused a checkpoint stall, so a large deployment should expect a short write-latency spike. DROP INDEX CONCURRENTLY IF EXISTS idx_feedback_records_tenant_updated_at; CREATE INDEX CONCURRENTLY idx_feedback_records_tenant_updated_at - ON feedback_records (tenant_id, updated_at); + ON feedback_records (tenant_id, updated_at, id); -- +goose down DROP INDEX CONCURRENTLY IF EXISTS idx_feedback_records_tenant_updated_at; diff --git a/openapi.yaml b/openapi.yaml index 6f6e99bc..bcd1c5a6 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -114,18 +114,22 @@ paths: - name: sort in: query description: | - Column to order by. Defaults to collected_at. Only columns that are non-null and - immutable after insert are offered: a mutable sort key would let a row move across - the cursor between pages and be silently skipped. + Column to order by. Defaults to collected_at. collected_at and created_at never + change. updated_at does, so it is accepted only with order=asc: ascending, a + record that changes while you page moves ahead of the cursor and is returned again + (upsert it by id), but never skipped; descending, it would land behind the cursor + and be lost, so that combination is a 400. Use sort=updated_at&order=asc with + updated_since for incremental extraction: every page is one ordered index range. schema: type: string default: collected_at enum: - collected_at - created_at + - updated_at - name: order in: query - description: Sort direction. Defaults to desc. Rows tied on the sort column are ordered by id ascending. + description: Sort direction. Defaults to desc, so sort=updated_at needs order=asc given explicitly. Rows tied on the sort column are ordered by id ascending. schema: type: string default: desc @@ -2730,16 +2734,15 @@ components: Filter by updated_at >= updated_since (ISO 8601, inclusive). updated_at is stamped on every write to a record — create, update, and every enrichment result (translation, sentiment, emotions) — so this selects records created OR changed since a point: the - parameter for incremental extraction. For the next run, use the time the previous run - started (its first response's Date header) minus an overlap, not the largest updated_at - received. The overlap must outlast the longest write to a record; Hub does not bound - that, so a few minutes is a default rather than a guarantee. Permanently deleted records - are never returned: periodically list every record, upsert what it returns (which also - repairs any change a slow write slipped past the overlap), and remove only local records - that were not returned and were created before that listing started (its first - response's Date header) minus the overlap. updated_at is not a sort key; keep sorting by - collected_at or created_at, which stays stable while matching records change. See - "Incremental extraction" in the filtering and sorting guide. + parameter for incremental extraction. Pair it with sort=updated_at&order=asc. For the + next run, use the time the previous run started (its first response's Date header) + minus an overlap, not the largest updated_at received. The overlap must outlast the + longest write to a record; Hub does not bound that, so a few minutes is a default + rather than a guarantee. Permanently deleted records are never returned: periodically + list every record, upsert what it returns (which also repairs any change a slow write + slipped past the overlap), and remove only local records that were not returned and + were created before that listing started (its first response's Date header) minus the + overlap. See "Incremental extraction" in the filtering and sorting guide. schema: type: string format: date-time diff --git a/tests/feedback_record_filters_test.go b/tests/feedback_record_filters_test.go index 50e3533e..53109bd5 100644 --- a/tests/feedback_record_filters_test.go +++ b/tests/feedback_record_filters_test.go @@ -337,57 +337,73 @@ func TestFeedbackRecordFilters_UpdatedSinceSelectsCreatedAndChanged(t *testing.T "updated_until bounds updated_at, so only the record nobody touched is at or before the cutoff") } -// TestFeedbackRecordFilters_UpdatedSincePaginationSurvivesUpdates is the reason updated_at is a -// filter and never a sort key. Records keep changing while a sync pages through them; ordering by -// an immutable column means an update moves a record's updated_at but never its position, so the -// traversal still returns every record exactly once. +// TestFeedbackRecordFilters_UpdatedSincePaginationSurvivesUpdates pages through an updated_since +// listing while records in it change, in every ordering a sync may use, and checks nothing is +// skipped. +// +// Each case touches the first record it was returned and the last one it has not reached yet: +// - created_at, either direction: the sort key does not move, so each record comes back once. +// - updated_at asc: both touched records move to the newest end, AHEAD of the cursor, so the +// already-returned one comes back a second time (the duplicate the docs tell callers to upsert) +// and the unreached one is still reached. This is the property that makes updated_at sortable +// ascending; descending it would land behind the cursor and be lost, which is why that +// combination is rejected rather than tested here. +// +// Each case seeds its own tenant: the touches change updated_at, so cases sharing records would +// see each other's edits and could pass for the wrong reason. func TestFeedbackRecordFilters_UpdatedSincePaginationSurvivesUpdates(t *testing.T) { - env := newFilterTestEnv(t, "updated-since-pages") - svc := service.NewFeedbackRecordsService( - repository.NewFeedbackRecordsRepository(env.db), nil, "", nil, nil, "", 0, "", - ) - - // Written before the cutoff, so it must never appear: without it this test would pass with - // updated_since ignored altogether. - before := env.seed(t) - cutoff := env.dbNow(t) - const records, pageLimit = 6, 2 - want := make(map[uuid.UUID]bool, records) - ordered := make([]uuid.UUID, 0, records) - - for range records { - record := env.seed(t) - want[record.ID] = true - ordered = append(ordered, record.ID) + cases := []struct { + sort models.SortField + order models.SortOrder + // timesFirstTouched is how often the record touched after being returned comes back. + timesFirstTouched int + }{ + {models.SortFieldCreatedAt, models.SortOrderAsc, 1}, + {models.SortFieldCreatedAt, models.SortOrderDesc, 1}, + {models.SortFieldUpdatedAt, models.SortOrderAsc, 2}, } - touch := func(id uuid.UUID) { - _, _, err := env.repo.Update(t.Context(), id, &models.UpdateFeedbackRecordRequest{ - UserID: new("touched-" + uuid.NewString()), - }) - require.NoError(t, err) - } + for _, testCase := range cases { + t.Run(string(testCase.sort)+" "+string(testCase.order), func(t *testing.T) { + env := newFilterTestEnv(t, "updated-since-pages") + svc := service.NewFeedbackRecordsService( + repository.NewFeedbackRecordsRepository(env.db), nil, "", nil, nil, "", 0, "", + ) + + // Written before the cutoff, so it must never appear: without it this test would pass + // with updated_since ignored altogether. + before := env.seed(t) + cutoff := env.dbNow(t) + + // Seeded one insert at a time, so created_at and updated_at both follow this order. + inOrder := make([]uuid.UUID, 0, records) + for range records { + inOrder = append(inOrder, env.seed(t).ID) + } - // Seeded in created_at order, so this is the traversal order for asc; desc walks it backwards. - for _, order := range []models.SortOrder{models.SortOrderAsc, models.SortOrderDesc} { - t.Run(string(order), func(t *testing.T) { - inOrder := slices.Clone(ordered) - if order == models.SortOrderDesc { + if testCase.order == models.SortOrderDesc { slices.Reverse(inOrder) } + touch := func(id uuid.UUID) { + _, _, err := env.repo.Update(t.Context(), id, &models.UpdateFeedbackRecordRequest{ + UserID: new("touched-" + uuid.NewString()), + }) + require.NoError(t, err) + } + seen := make(map[uuid.UUID]int, records) nextCursor := "" - for page := range records { + for page := range records + 2 { tenant := env.tenant resp, err := svc.ListFeedbackRecords(t.Context(), &models.ListFeedbackRecordsFilters{ TenantID: &tenant, UpdatedSince: &cutoff, - Sort: models.SortFieldCreatedAt, - Order: order, + Sort: testCase.sort, + Order: testCase.order, Limit: pageLimit, Cursor: nextCursor, }) @@ -398,11 +414,7 @@ func TestFeedbackRecordFilters_UpdatedSincePaginationSurvivesUpdates(t *testing. } if page == 0 { - // Change the first record returned and the last one not reached yet. Their - // created_at, the sort key, does not move, so each is still returned once. Were - // updated_at the sort key, both would move to the newest end of the ordering: under - // asc the first would come round again (returned twice), and under desc the last - // would land behind the cursor (never returned). + require.Equal(t, inOrder[0], resp.Data[0].ID, "the first page must start where the seed order says") touch(inOrder[0]) touch(inOrder[records-1]) } @@ -415,12 +427,17 @@ func TestFeedbackRecordFilters_UpdatedSincePaginationSurvivesUpdates(t *testing. require.Empty(t, nextCursor, "pagination did not terminate") assert.NotContains(t, seen, before.ID, "a record last written before updated_since must not be returned") - assert.Len(t, seen, records, "every record must be returned") - for id, times := range seen { - assert.True(t, want[id], "record %s does not belong to this traversal", id) - assert.Equal(t, 1, times, "record %s returned %d times", id, times) + for _, id := range inOrder { + want := 1 + if id == inOrder[0] { + want = testCase.timesFirstTouched + } + + assert.Equal(t, want, seen[id], "record %s returned %d times, want %d", id, seen[id], want) } + + assert.Len(t, seen, records, "no record outside the traversal may appear") }) } } @@ -811,6 +828,7 @@ func TestFeedbackRecordFilters_PaginationInvariant(t *testing.T) { {models.SortFieldCollectedAt, models.SortOrderAsc}, {models.SortFieldCreatedAt, models.SortOrderDesc}, {models.SortFieldCreatedAt, models.SortOrderAsc}, + {models.SortFieldUpdatedAt, models.SortOrderAsc}, } for _, ordering := range orderings { From 70bb1cc1d41ff30f8a8599f49f908980d986e71d Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Fri, 9 Oct 2026 09:03:56 +0000 Subject: [PATCH 13/15] docs: name the Airbyte refresh that repairs a cursor sync (ENG-3420) Only a refresh that clears the destination (refresh and remove records, or Full refresh | Overwrite) repairs late commits, out-of-order versions and deletes; a retaining refresh keeps the larger-updated_at version. Refreshes are triggered, not scheduled. Also: the parity test now asserts count leaves the four list-only params undocumented, and the exec-mode comment says one round trip only once the description is cached. --- docs/src/content/docs/guides/hub-airbyte.mdx | 8 ++++++-- .../models/feedback_records_filters_openapi_test.go | 12 +++++++++--- internal/repository/feedback_records_repository.go | 3 ++- 3 files changed, 17 insertions(+), 6 deletions(-) diff --git a/docs/src/content/docs/guides/hub-airbyte.mdx b/docs/src/content/docs/guides/hub-airbyte.mdx index fe1cc295..4f833771 100644 --- a/docs/src/content/docs/guides/hub-airbyte.mdx +++ b/docs/src/content/docs/guides/hub-airbyte.mdx @@ -220,7 +220,7 @@ Other supported destinations include **Postgres**, **ClickHouse**, **S3** (raw f @@ -229,7 +229,11 @@ Other supported destinations include **Postgres**, **ClickHouse**, **S3** (raw f when a write *starts*, so a write that commits after a sync has read past it can carry an earlier timestamp than the cursor and is skipped until the record changes again. Two writes to one record can also commit out of order, so the destination may keep an older version. Deleted records are - never replicated. Schedule a periodic full refresh of the stream to repair all three, and see + never replicated. Only a refresh that clears the destination first repairs all three: Airbyte's + **refresh and remove records** (or a **Full refresh \| Overwrite** sync). A refresh that retains + records does not, because deduping keeps the version with the largest `updated_at`, and deleted rows + stay. Refreshes are not part of the connection schedule; trigger one periodically from the UI, the + Airbyte API or your orchestrator. See [Incremental extraction](/core-concepts/filtering-and-sorting#incremental-extraction) for the procedure the Hub API supports. diff --git a/internal/models/feedback_records_filters_openapi_test.go b/internal/models/feedback_records_filters_openapi_test.go index 4099ed0c..42ccf9e4 100644 --- a/internal/models/feedback_records_filters_openapi_test.go +++ b/internal/models/feedback_records_filters_openapi_test.go @@ -180,9 +180,11 @@ type openAPIOperationParameters struct { // `form:"updated_untl"` would pass every other test while the decoder silently ignored the real // parameter — for updated_since (ENG-3420), a sync that reads the whole tenant every run. // -// List must document every field; count every field except the four pagination and ordering ones, -// which it does not accept. Every documented parameter must in turn decode into a field, and every -// timestamp field must be documented as date-time. +// List must document every field. Count shares the struct but documents only the filters: the four +// pagination and ordering fields mean nothing to a count, so the spec keeps them off it (the handler +// still decodes and validates them, which is why an out-of-range limit on count is a 400). Every +// documented parameter must in turn decode into a field, and every timestamp field must be +// documented as date-time. func TestOpenAPIDocumentsEveryFilterByItsFormName(t *testing.T) { t.Parallel() @@ -244,6 +246,10 @@ func TestOpenAPIDocumentsEveryFilterByItsFormName(t *testing.T) { formNames[name] = true if operation.skipNames[name] { + if _, documented := operation.formats[name]; documented { + t.Fatalf("%s documents %q, a list-only parameter", operation.path, name) + } + continue } diff --git a/internal/repository/feedback_records_repository.go b/internal/repository/feedback_records_repository.go index ed048577..91764865 100644 --- a/internal/repository/feedback_records_repository.go +++ b/internal/repository/feedback_records_repository.go @@ -662,7 +662,8 @@ func scanBackfillTargetIDs(rows pgx.Rows, name string) ([]uuid.UUID, error) { // the tenant's ordering index that discards nearly everything: under 1 ms for the first five calls, // ~15 ms from the sixth, on a local 200k-row tenant. Every filter on these endpoints is skewed by // tenant size the same way, so the mode covers list, cursor pages and count alike. The cost is a -// plan per execution, still in one round trip because the statement description stays cached. +// plan per execution; it stays one round trip once pgx has cached the statement's description +// (the first execution of each query text on a connection takes two, as with the default mode). func filterQueryArgs(args []any) []any { return append([]any{pgx.QueryExecModeCacheDescribe}, args...) } From 2982598d9d5ec19d6a3b10a5d2fd2d901deca012 Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Fri, 9 Oct 2026 09:25:05 +0000 Subject: [PATCH 14/15] docs: qualify the ascending updated_at guarantee and bound runs (ENG-3420) Ascending normally moves a changed record ahead of the cursor, but a write that started before the cursor's position and commits after it still lands behind; the next run's run-start watermark recovers it. Say so in the spec, the SortField invariant and the guide, restore the created_at-sort reason for the watermark, and recommend updated_until = run start so a run stops chasing its own write wave. The asc-only rule no longer double-reports an order its oneof tag already rejects. --- .../core-concepts/filtering-and-sorting.mdx | 26 ++++++++++++++----- internal/api/validation/validation.go | 7 +++-- internal/api/validation/validation_test.go | 17 ++++++++++-- internal/models/feedback_records_filters.go | 10 ++++--- openapi.yaml | 12 ++++++--- 5 files changed, 55 insertions(+), 17 deletions(-) diff --git a/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx b/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx index 09b8f080..1f5a6c29 100644 --- a/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx +++ b/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx @@ -137,6 +137,7 @@ curl -G "$HUB/v1/feedback-records" \ -H "Authorization: Bearer $HUB_API_KEY" \ -d tenant_id=org-123 \ -d updated_since=2026-10-05T07:55:00Z \ + -d updated_until=2026-10-05T09:00:00Z \ -d sort=updated_at \ -d order=asc \ -d limit=1000 @@ -148,10 +149,21 @@ sorting by `created_at` also works, but the database cannot tell how few records read the whole tenant for every page. Ascending is the only direction `updated_at` can be sorted in, because `updated_at` changes while you -page. A record that changes mid-run gets a newer `updated_at`, which moves it *ahead* of the cursor: -it comes back later in the same run, and you upsert it again. Descending would move it *behind* the -cursor, where it would be skipped, so `sort=updated_at` with `order=desc` — or with no `order`, which -defaults to `desc` — is a `400`. +page. A record that changes mid-run normally gets a newer `updated_at`, which moves it *ahead* of the +cursor: it comes back later in the same run, and you upsert it again. Descending would move every such +record *behind* the cursor, where it would be skipped, so `sort=updated_at` with `order=desc` — or with +no `order`, which defaults to `desc` — is a `400`. + +Ascending still has one exception: `updated_at` is when a write *started*, so a write that started +before the cursor's position and commits after the run has read past it lands behind the cursor, as +can the earlier of two writes to one record that commit out of order. Those are missing from this run +and picked up by the next one, as long as you choose the next `updated_since` as below. + +**Bound each run with `updated_until`.** Set it to the run's start time (the first response's `Date` +header). Otherwise a run never quite ends while records keep changing: every write during the run +moves a record ahead of the cursor, and a new record collects several (translation, sentiment, +emotions) in its first minutes. Anything that changes after the start has a newer `updated_at` and +belongs to the next run anyway. ### Choosing the next `updated_since` @@ -159,8 +171,10 @@ defaults to `desc` — is a `400`. Do not use the largest `updated_at` you received. `updated_at` is when a write *started*, and a write becomes visible only when it commits: one that was still in flight when your run read past its timestamp is missing from this run, and its `updated_at` is older than the newest one you received. -A watermark taken from the results skips it for good. A watermark taken from the run's start, minus -the overlap below, reads it next time. +If you sort by `created_at` instead, there is a second way to lose a change: a record you paged past +early can change while later pages are read, and its new `updated_at` is still older than the newest +one on your last page. Either way, a watermark taken from the results skips the change for good; a +watermark taken from the run's start, minus the overlap below, reads it next time. - **Take the start time from the server, not from your own clock.** Every response carries an HTTP `Date` header; use the one on the run's first response, so drift between your clock and the diff --git a/internal/api/validation/validation.go b/internal/api/validation/validation.go index 7e939bbc..b8527c31 100644 --- a/internal/api/validation/validation.go +++ b/internal/api/validation/validation.go @@ -182,8 +182,11 @@ func validateListFeedbackRecordsFilters(structLevel validator.StructLevel) { structLevel.ReportError(inverted.MinValue, inverted.MinParam, inverted.StructField, rangeBoundsTag, inverted.MaxParam) } - // An omitted order defaults to desc, so sort=updated_at needs order=asc spelled out. - if filters.Sort == models.SortFieldUpdatedAt && filters.Order != models.SortOrderAsc { + // An omitted order defaults to desc, so sort=updated_at needs order=asc spelled out. An order that + // is not asc/desc at all is already reported by its oneof tag; repeating it here would give the + // caller two reasons for one parameter. + if filters.Sort == models.SortFieldUpdatedAt && + (filters.Order == "" || filters.Order == models.SortOrderDesc) { structLevel.ReportError(filters.Order, "order", "Order", updatedAtAscendingTag, "") } } diff --git a/internal/api/validation/validation_test.go b/internal/api/validation/validation_test.go index 489be461..68ce10c0 100644 --- a/internal/api/validation/validation_test.go +++ b/internal/api/validation/validation_test.go @@ -192,8 +192,6 @@ func TestDecodeEnumSliceQueryParams(t *testing.T) { }) } -// TestValidateInvertedRangeFilters verifies an inverted min/max pair is a 400 naming the lower -// bound, rather than an empty result page the caller reads as "there is no such feedback". // TestValidateUpdatedAtSortIsAscendingOnly covers the struct-level rule that sort=updated_at needs // order=asc spelled out: descending would skip records that change mid-traversal, and an omitted // order defaults to desc. The 400 names order, the parameter the client has to change. @@ -226,8 +224,23 @@ func TestValidateUpdatedAtSortIsAscendingOnly(t *testing.T) { TenantID: &tenant, Sort: models.SortFieldCreatedAt, })) }) + + // An unrecognised order is the oneof tag's to report; one parameter gets one reason. + t.Run("does not double-report an invalid order", func(t *testing.T) { + err := ValidateStruct(models.ListFeedbackRecordsFilters{ + TenantID: &tenant, Sort: models.SortFieldUpdatedAt, Order: models.SortOrder("ASC"), + }) + require.Error(t, err) + + var validationErrs validator.ValidationErrors + require.ErrorAs(t, err, &validationErrs) + require.Len(t, validationErrs, 1) + assert.Equal(t, "oneof", validationErrs[0].Tag()) + }) } +// TestValidateInvertedRangeFilters verifies an inverted min/max pair is a 400 naming the lower +// bound, rather than an empty result page the caller reads as "there is no such feedback". func TestValidateInvertedRangeFilters(t *testing.T) { tenant := "org-123" diff --git a/internal/models/feedback_records_filters.go b/internal/models/feedback_records_filters.go index 49f9fa9f..99dc57a3 100644 --- a/internal/models/feedback_records_filters.go +++ b/internal/models/feedback_records_filters.go @@ -15,10 +15,14 @@ import ( // Every member must also be immutable after insert, with one deliberate exception. A mutable sort // key lets a row move across the cursor between pages. collected_at and created_at never change. // updated_at does — every PATCH and enrichment write bumps it — and is allowed in ONE direction -// only (ENG-3420): ascending, a changed row moves AHEAD of the cursor, so it is returned again -// (a duplicate the caller upserts away) but never skipped. Descending it would move behind the +// only (ENG-3420): ascending, a changed row normally moves AHEAD of the cursor, so it is returned +// again (a duplicate the caller upserts away). Descending, every such row would move behind the // cursor and be lost, which is why resolveListOrdering and the HTTP validator both reject -// sort=updated_at with any order but asc. It exists because filtering on updated_at while ordering +// sort=updated_at with any order but asc. Ascending is not airtight either: updated_at is the +// writing transaction's NOW(), so a write that started before the cursor's position and commits +// after it — or the earlier of two writes to one row that commit out of order — lands behind the +// cursor. The documented sync recovers those on the next run, whose updated_since is the previous +// run's start minus an overlap longer than any write. It exists because filtering on updated_at while ordering // by another column leaves the planner guessing how many rows match — it multiplies the tenant's // share by the window's share and cannot see that recent activity belongs to other tenants — and // a wrong guess walks the whole tenant per page. Ordered by (updated_at, id) on the diff --git a/openapi.yaml b/openapi.yaml index bcd1c5a6..3a660085 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -116,10 +116,14 @@ paths: description: | Column to order by. Defaults to collected_at. collected_at and created_at never change. updated_at does, so it is accepted only with order=asc: ascending, a - record that changes while you page moves ahead of the cursor and is returned again - (upsert it by id), but never skipped; descending, it would land behind the cursor - and be lost, so that combination is a 400. Use sort=updated_at&order=asc with - updated_since for incremental extraction: every page is one ordered index range. + record that changes while you page normally moves ahead of the cursor and is + returned again (upsert it by id); descending, every such record would land behind + the cursor and be lost, so that combination is a 400. Ascending still misses a + write that started before the cursor's position and commits after it; the next + run picks it up when its updated_since is the previous run's start minus an + overlap. Use sort=updated_at&order=asc with updated_since (and updated_until set + to the run's start) for incremental extraction: every page is one ordered index + range. schema: type: string default: collected_at From d164a02d1d94ecff13ef6ad0487336368ffbe596 Mon Sep 17 00:00:00 2001 From: Tiago Farto Date: Fri, 9 Oct 2026 09:34:34 +0000 Subject: [PATCH 15/15] docs: take both sync bounds from one Date header (ENG-3420) updated_until has to be set before the first page, and a client that used its own clock for it could fall short of the next run's updated_since and leave records neither run reads. Start each run with a limit=1 request and use its Date as this run's updated_until and, minus the overlap, the next run's updated_since. --- .../docs/core-concepts/filtering-and-sorting.mdx | 13 ++++++++----- openapi.yaml | 7 ++++--- 2 files changed, 12 insertions(+), 8 deletions(-) diff --git a/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx b/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx index 1f5a6c29..b9a373c7 100644 --- a/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx +++ b/docs/src/content/docs/core-concepts/filtering-and-sorting.mdx @@ -159,11 +159,14 @@ before the cursor's position and commits after the run has read past it lands be can the earlier of two writes to one record that commit out of order. Those are missing from this run and picked up by the next one, as long as you choose the next `updated_since` as below. -**Bound each run with `updated_until`.** Set it to the run's start time (the first response's `Date` -header). Otherwise a run never quite ends while records keep changing: every write during the run -moves a record ahead of the cursor, and a new record collects several (translation, sentiment, -emotions) in its first minutes. Anything that changes after the start has a newer `updated_at` and -belongs to the next run anyway. +**Bound each run with `updated_until`.** Otherwise a run never quite ends while records keep +changing: every write during the run moves a record ahead of the cursor, and a new record collects +several (translation, sentiment, emotions) in its first minutes. Start each run with a cheap request +(a page with `limit=1`) and read its `Date` header: that one value, *S*, is this run's +`updated_until`, and *S* minus the overlap is the next run's `updated_since`. Both bounds have to come +from the same *S* — an `updated_until` from your own clock can fall short of the next run's +`updated_since` and leave records in between that neither run ever reads. Anything that changes after +*S* has a newer `updated_at` and belongs to the next run. ### Choosing the next `updated_since` diff --git a/openapi.yaml b/openapi.yaml index 3a660085..4ddff4f3 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -121,9 +121,10 @@ paths: the cursor and be lost, so that combination is a 400. Ascending still misses a write that started before the cursor's position and commits after it; the next run picks it up when its updated_since is the previous run's start minus an - overlap. Use sort=updated_at&order=asc with updated_since (and updated_until set - to the run's start) for incremental extraction: every page is one ordered index - range. + overlap. Use sort=updated_at&order=asc with updated_since and updated_until for + incremental extraction, both derived from the Date header of one request made + at the run's start (see the filtering and sorting guide): every page is one + ordered index range. schema: type: string default: collected_at