From 2acacc87a700dd8cc8635647697ef3f211b73646 Mon Sep 17 00:00:00 2001 From: Matthew Meszaros Date: Wed, 14 Oct 2026 12:39:53 +0200 Subject: [PATCH] feat: refresh the shared warmup seven-day snapshot on a detached context while serving the previous one, and derive the Emails tab status chunk layout from committed state instead of writing a ref during render --- .github/workflows/ci.yml | 4 +- internal/repository/pg_warmup_partner_load.go | 63 +++++++++++++++---- .../warmup_partner_load_live_test.go | 58 +++++++++++++++++ .../hooks/app/analytics/useAccountStatuses.ts | 15 +++-- 4 files changed, 121 insertions(+), 19 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7805a92d7..214865289 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -207,8 +207,8 @@ jobs: run: | set -o pipefail go run ./cmd/migrate - go test -race -v -p 1 ./internal/repository ./internal/app/analytics -run '^TestLive(WarmupPartnerCountsMatchIndividualCandidates|WarmupSummariesMatchIndividualReads|LinkedMailboxStatusBatchPreservesDetails|WarmupPartnerWeekCacheMatchesExactReadAndKeepsTheCapLive|WarmupPartnerCandidatesCarryReciprocityAndHonourTheCap|AccountStatusPageMatchesSingleReadsInBoundedQueries)$' -count=1 | tee linked-mailbox-batch.log - for test in TestLiveWarmupPartnerCountsMatchIndividualCandidates TestLiveWarmupSummariesMatchIndividualReads TestLiveLinkedMailboxStatusBatchPreservesDetails TestLiveWarmupPartnerWeekCacheMatchesExactReadAndKeepsTheCapLive TestLiveWarmupPartnerCandidatesCarryReciprocityAndHonourTheCap TestLiveAccountStatusPageMatchesSingleReadsInBoundedQueries; do + go test -race -v -p 1 ./internal/repository ./internal/app/analytics -run '^TestLive(WarmupPartnerCountsMatchIndividualCandidates|WarmupSummariesMatchIndividualReads|LinkedMailboxStatusBatchPreservesDetails|WarmupPartnerWeekCacheMatchesExactReadAndKeepsTheCapLive|WarmupPartnerWeekCacheRefreshesDetachedAndServesStale|WarmupPartnerCandidatesCarryReciprocityAndHonourTheCap|AccountStatusPageMatchesSingleReadsInBoundedQueries)$' -count=1 | tee linked-mailbox-batch.log + for test in TestLiveWarmupPartnerCountsMatchIndividualCandidates TestLiveWarmupSummariesMatchIndividualReads TestLiveLinkedMailboxStatusBatchPreservesDetails TestLiveWarmupPartnerWeekCacheMatchesExactReadAndKeepsTheCapLive TestLiveWarmupPartnerWeekCacheRefreshesDetachedAndServesStale TestLiveWarmupPartnerCandidatesCarryReciprocityAndHonourTheCap TestLiveAccountStatusPageMatchesSingleReadsInBoundedQueries; do grep -q -- "--- PASS: $test " linked-mailbox-batch.log done if grep -q -- '--- SKIP:' linked-mailbox-batch.log; then exit 1; fi diff --git a/internal/repository/pg_warmup_partner_load.go b/internal/repository/pg_warmup_partner_load.go index 57dd50907..a4a483529 100644 --- a/internal/repository/pg_warmup_partner_load.go +++ b/internal/repository/pg_warmup_partner_load.go @@ -22,6 +22,10 @@ type partnerWeek struct { sent, received, junked int } +// partnerWeekRefreshTimeout bounds one pool-wide refresh, which runs detached +// from the caller that started it. +const partnerWeekRefreshTimeout = 30 * time.Second + // partnerWeekCache holds the pool-wide seven-day counts for one process, so // every sender shares one aggregate instead of rescanning the window per call. type partnerWeekCache struct { @@ -29,6 +33,8 @@ type partnerWeekCache struct { mu sync.Mutex at time.Time data map[uuid.UUID]partnerWeek + // refreshing is closed when the refresh in flight ends; nil when none is. + refreshing chan struct{} } func newPartnerWeekCache(ttl time.Duration) *partnerWeekCache { @@ -61,25 +67,56 @@ var ( ) // partnerWeeks returns the seven-day counts for ids: from the shared cache -// when the repository has one, otherwise read exactly for those ids. +// when the repository has one, otherwise read exactly for those ids. An +// expired snapshot is served while one refresh replaces it, so no sender waits +// on the pool-wide scan once the first one has landed. func (r *warmupRepository) partnerWeeks(ctx context.Context, ids []uuid.UUID) (map[uuid.UUID]partnerWeek, error) { - if c := r.partnerWeeks7d; c != nil { - c.mu.Lock() - defer c.mu.Unlock() - if c.data != nil && time.Since(c.at) < c.ttl { - return c.data, nil - } - // The pool, not a caller's transaction: the snapshot is shared. - data, err := scanPartnerWeeks(ctx, r.db, partnerWeekAllSQL) - if err != nil { - return nil, err - } - c.data, c.at = data, time.Now() + c := r.partnerWeeks7d + if c == nil { + return scanPartnerWeeks(ctx, resultDB(ctx, r.db), partnerWeekIDsSQL, ids) + } + c.mu.Lock() + data, fresh := c.data, c.data != nil && time.Since(c.at) < c.ttl + wait := c.refreshing + if !fresh && wait == nil { + wait = make(chan struct{}) + c.refreshing = wait + go r.refreshPartnerWeeks(context.WithoutCancel(ctx), c, wait) + } + c.mu.Unlock() + if data != nil { return data, nil } + select { + case <-wait: + case <-ctx.Done(): + return nil, ctx.Err() + } + c.mu.Lock() + data = c.data + c.mu.Unlock() + if data != nil { + return data, nil + } + // The first refresh failed; this caller reads its own ids exactly. return scanPartnerWeeks(ctx, resultDB(ctx, r.db), partnerWeekIDsSQL, ids) } +// refreshPartnerWeeks replaces the snapshot and keeps the old one on failure. +func (r *warmupRepository) refreshPartnerWeeks(ctx context.Context, c *partnerWeekCache, done chan struct{}) { + ctx, cancel := context.WithTimeout(ctx, partnerWeekRefreshTimeout) + defer cancel() + // The pool, not a caller's transaction: the snapshot is shared. + data, err := scanPartnerWeeks(ctx, r.db, partnerWeekAllSQL) + c.mu.Lock() + if err == nil { + c.data, c.at = data, time.Now() + } + c.refreshing = nil + c.mu.Unlock() + close(done) +} + func scanPartnerWeeks(ctx context.Context, db sendResultDB, query string, args ...any) (map[uuid.UUID]partnerWeek, error) { rows, err := db.Query(ctx, query, args...) if err != nil { diff --git a/internal/repository/warmup_partner_load_live_test.go b/internal/repository/warmup_partner_load_live_test.go index 5e92d8e7c..be82faaf8 100644 --- a/internal/repository/warmup_partner_load_live_test.go +++ b/internal/repository/warmup_partner_load_live_test.go @@ -2,6 +2,7 @@ package repository import ( "context" + "reflect" "testing" "time" @@ -109,3 +110,60 @@ func TestPartnerInboundCapMatchesTheSQLRule(t *testing.T) { } } } + +// A caller that gives up does not cancel the shared refresh, and an expired +// snapshot is served while one refresh replaces it. +func TestLiveWarmupPartnerWeekCacheRefreshesDetachedAndServesStale(t *testing.T) { + f := newPartnerOrgFixture(t) + c := newPartnerWeekCache(time.Hour) + repo := &warmupRepository{db: f.pool, partnerWeeks7d: c} + ids := []uuid.UUID{f.sender, f.sibling, f.outside} + + gone, cancel := context.WithCancel(context.Background()) + cancel() + _, _ = repo.partnerWeeks(gone, ids) + got, err := repo.partnerWeeks(context.Background(), ids) + if err != nil { + t.Fatal(err) + } + c.mu.Lock() + landed := c.data != nil + c.mu.Unlock() + if !landed { + t.Fatal("the refresh a cancelled caller started did not land") + } + exact, err := (&warmupRepository{db: f.pool}).partnerWeeks(context.Background(), ids) + if err != nil { + t.Fatal(err) + } + for _, id := range ids { + if got[id] != exact[id] { + t.Fatalf("%s: snapshot %+v, exact %+v", id, got[id], exact[id]) + } + } + + c.mu.Lock() + c.at = time.Now().Add(-2 * time.Hour) + before := c.data + c.mu.Unlock() + stale, err := repo.partnerWeeks(context.Background(), ids) + if err != nil { + t.Fatal(err) + } + if reflect.ValueOf(stale).Pointer() != reflect.ValueOf(before).Pointer() { + t.Fatal("an expired snapshot was not served while it refreshed") + } + deadline := time.Now().Add(10 * time.Second) + for { + c.mu.Lock() + refreshed := time.Since(c.at) < time.Minute && c.refreshing == nil + c.mu.Unlock() + if refreshed { + break + } + if time.Now().After(deadline) { + t.Fatal("the expired snapshot was never replaced") + } + time.Sleep(10 * time.Millisecond) + } +} diff --git a/web/src/lib/api/hooks/app/analytics/useAccountStatuses.ts b/web/src/lib/api/hooks/app/analytics/useAccountStatuses.ts index 9ebeb9692..bfd442504 100644 --- a/web/src/lib/api/hooks/app/analytics/useAccountStatuses.ts +++ b/web/src/lib/api/hooks/app/analytics/useAccountStatuses.ts @@ -1,4 +1,4 @@ -import { useMemo, useRef } from "react"; +import { useMemo, useState } from "react"; import { useQueries, type UseQueryResult } from "@tanstack/react-query"; import getAccountStatuses from "@/lib/api/client/app/analytics/getAccountStatuses"; import type AccountStatus from "@/lib/api/models/app/analytics/AccountStatus"; @@ -36,14 +36,21 @@ export default function useAccountStatuses(emailIds: string[], options?: { enabl // list never re-keys (and refetches) the chunks already loaded. const enabled = options?.enabled ?? true; const sourceRevision = options?.sourceRevision ?? 0; - const chunksRef = useRef([]); + // The layout is state derived from the last committed one, so a render + // React discards never shapes the next render's keys. + const [layout, setLayout] = useState<{ ids: string[]; chunks: string[][] }>(() => ({ ids: emailIds, chunks: stableChunks([], emailIds) })); + let chunks = layout.chunks; + if (layout.ids !== emailIds) { + chunks = stableChunks(layout.chunks, emailIds); + setLayout({ ids: emailIds, chunks }); + } const queries = useMemo( - () => (chunksRef.current = stableChunks(chunksRef.current, emailIds)).map((ids) => ({ + () => chunks.map((ids) => ({ queryKey: ["analytics", "accounts", "list", ids, sourceRevision], queryFn: ({ signal }: { signal: AbortSignal }) => getAccountStatuses(ids, signal), enabled: ids.length > 0 && enabled, })), - [emailIds, enabled, sourceRevision], + [chunks, enabled, sourceRevision], ); return useQueries({ queries, combine: combineAccountStatuses });