mirror of
https://github.com/warmbly/warmbly.git
synced 2026-11-04 08:15:38 +00:00
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
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<string[][]>([]);
|
||||
// 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 });
|
||||
|
||||
Reference in New Issue
Block a user