From d4482e1c4421ba67f237d7f8862c9ae9a2afe5ef Mon Sep 17 00:00:00 2001 From: chrisnojima Date: Sun, 26 Jul 2026 15:01:32 -0400 Subject: [PATCH 1/8] fix(search): converge the chat index instead of re-fetching it forever reindexConv paged over a conv's raw min..max message range rather than over the ids actually missing, so every pass re-fetched everything already indexed - 57,648 already-seen messages re-added in a single pass - while never reaching the top of a large conv, because each pass restarts at the lowest missing id and stops once the inbox-wide budget is spent. And a conv can never reach numMissing 0 while ids the server will never return (deleted, or never delivered) keep counting as missing, so SelectiveSync re-queued it every interval forever: ~1800 remote getMessages every 5 minutes on an idle desktop client, with numMissing byte-identical across 7 consecutive cycles. Pages now walk the missing ids and MarkSeen accounts for the ids the fetch will never return. A conv that still cannot converge is backed off with a bounded exponential skip rather than suppressed - "numMissing did not move" is strong evidence, not proof, and a budgeted pass can cover only unfetchable ids while real work remains further up the range. The clamp is on the shift exponent, not the result: streak is unbounded and a large enough shift overflows int negative and then to 0, which reads as "no skips" and turns the backoff off for good. Underneath that sat a worse class of bug: several paths published "these messages are indexed" without the tokens that make them findable ever reaching disk. Writes evicted from the cache before a flush, a flush that failed partway and dropped what it had not written, Remove writing metadata straight to disk ahead of its pending tokens, Add marking a message before indexing it, deletes racing the flush's write-outside-the-lock window, and a storage op whose callback reported success when the op had failed or never run. Each leaves a conv reading as fully indexed with messages no search will return, and nothing revisits a conv that reads as complete. Deletes are now queued in the pending set as nil entries and applied in order, failed writes are requeued, and the storage callback carries an error so the caller stops rather than marking. Because such a conv cannot repair itself, the index has to be dropped once on upgrade. indexMetadataVersion 3 -> 4 does that. Bumping indexVersion instead would have taken the token and alias entries with it and left inbox search returning nothing until the rebuild finished; the damage is confined to the metadata, so every conv re-indexes onto the entries already on disk and search keeps working throughout. Two things survive that a full rebuild would have cleared: alias refcounts, inflated by the re-add, and token entries naming since-deleted messages, which are filtered when a search resolves them and so cost a lookup rather than a wrong result. maxDirtyEntries was 1000, below the ~2600 entries one 300-message page produces, so it fired on every batch: 65 flushes in 2.9s with a 0.04s median gap, re-encrypting hot token entries whole on every pass. It is a memory bound, not a correctness one, so it belongs above one batch. At 20000: 7 flushes over a comparable window, and time spent flushing fell from ~69% of wall clock to 40%. Every one of these is covered by a test that fails against the pre-fix code. --- go/chat/search/flush_test.go | 596 +++++++++++++++++++++++++++++++++++ go/chat/search/indexer.go | 337 ++++++++++++++++---- go/chat/search/md.go | 17 +- go/chat/search/stall_test.go | 241 ++++++++++++++ go/chat/search/storage.go | 402 ++++++++++++++++++----- 5 files changed, 1456 insertions(+), 137 deletions(-) create mode 100644 go/chat/search/flush_test.go create mode 100644 go/chat/search/stall_test.go diff --git a/go/chat/search/flush_test.go b/go/chat/search/flush_test.go new file mode 100644 index 000000000000..d14295367ffa --- /dev/null +++ b/go/chat/search/flush_test.go @@ -0,0 +1,596 @@ +package search + +import ( + "context" + "fmt" + "sync" + "testing" + + "github.com/keybase/client/go/chat/globals" + "github.com/keybase/client/go/externalstest" + "github.com/keybase/client/go/protocol/chat1" + "github.com/keybase/client/go/protocol/gregor1" + "github.com/stretchr/testify/require" +) + +// The caches are bounded LRUs (3000 tokens, 10000 aliases, 500 metadata) and a +// heavy index pass writes far more than that between flushes, so a dirty entry +// being evicted before it reaches disk is the normal case, not a corner one. +// Purging in these tests stands in for that eviction: it leaves the entry dirty +// but no longer resident, which is the state eviction produces. +// +// The real disk store encrypts with the device key and so needs a logged-in +// user; this fake keeps the tests to the contract that matters, which is what +// Flush wrote and what a read returns after an eviction. +type memDiskStorage struct { + sync.Mutex + tokens map[string]*tokenEntry + aliases map[string]*aliasEntry + md map[chat1.ConvIDStr]*indexMetadata + + // set to make the corresponding call fail, so the error paths - which are + // where entries get lost - are reachable from a test + failTokenPut error + failAliasPut error + failMdPut error + failTokenGet error + + // runs once, during a PutTokenEntry, to interleave an operation with a + // flush's disk writes - which happen outside the store lock + onPutToken func() +} + +func newMemDiskStorage() *memDiskStorage { + return &memDiskStorage{ + tokens: make(map[string]*tokenEntry), + aliases: make(map[string]*aliasEntry), + md: make(map[chat1.ConvIDStr]*indexMetadata), + } +} + +func (m *memDiskStorage) key(convID chat1.ConversationID, token string) string { + return convID.ConvIDStr().String() + ":" + token +} + +func (m *memDiskStorage) GetTokenEntry(ctx context.Context, convID chat1.ConversationID, + token string, +) (*tokenEntry, error) { + m.Lock() + defer m.Unlock() + if m.failTokenGet != nil { + return nil, m.failTokenGet + } + return m.tokens[m.key(convID, token)], nil +} + +func (m *memDiskStorage) PutTokenEntry(ctx context.Context, convID chat1.ConversationID, + token string, te *tokenEntry, +) error { + m.Lock() + defer m.Unlock() + if m.failTokenPut != nil { + return m.failTokenPut + } + if te == nil { + return fmt.Errorf("PutTokenEntry called with a nil entry: a delete must not be written as a value") + } + if m.onPutToken != nil { + hook := m.onPutToken + m.onPutToken = nil + m.Unlock() + hook() + m.Lock() + } + m.tokens[m.key(convID, token)] = te + return nil +} + +func (m *memDiskStorage) RemoveTokenEntry(ctx context.Context, convID chat1.ConversationID, token string) { + m.Lock() + defer m.Unlock() + delete(m.tokens, m.key(convID, token)) +} + +func (m *memDiskStorage) GetAliasEntry(ctx context.Context, alias string) (*aliasEntry, error) { + m.Lock() + defer m.Unlock() + return m.aliases[alias], nil +} + +func (m *memDiskStorage) PutAliasEntry(ctx context.Context, alias string, ae *aliasEntry) error { + m.Lock() + defer m.Unlock() + if m.failAliasPut != nil { + return m.failAliasPut + } + if ae == nil { + return fmt.Errorf("PutAliasEntry called with a nil entry: a delete must not be written as a value") + } + m.aliases[alias] = ae + return nil +} + +func (m *memDiskStorage) RemoveAliasEntry(ctx context.Context, alias string) { + m.Lock() + defer m.Unlock() + delete(m.aliases, alias) +} + +func (m *memDiskStorage) GetMetadata(ctx context.Context, convID chat1.ConversationID) (*indexMetadata, error) { + m.Lock() + defer m.Unlock() + return m.md[convID.ConvIDStr()], nil +} + +func (m *memDiskStorage) PutMetadata(ctx context.Context, convID chat1.ConversationID, md *indexMetadata) error { + m.Lock() + defer m.Unlock() + if m.failMdPut != nil { + return m.failMdPut + } + m.md[convID.ConvIDStr()] = md + return nil +} + +func (m *memDiskStorage) Clear(ctx context.Context, uid gregor1.UID, convID chat1.ConversationID) error { + m.Lock() + defer m.Unlock() + delete(m.md, convID.ConvIDStr()) + return nil +} + +func setupFlushTestStore(t *testing.T, label string) (context.Context, *store, *memDiskStorage, chat1.ConversationID) { + tc := externalstest.SetupTest(t, label, 2) + t.Cleanup(tc.Cleanup) + g := globals.NewContext(tc.G, &globals.ChatContext{}) + s := newStore(g, gregor1.UID([]byte{1, 2, 3, 4})) + disk := newMemDiskStorage() + s.diskStorage = disk + convID := chat1.ConversationID([]byte{ + 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, + }) + return context.TODO(), s, disk, convID +} + +func TestFlushWritesEvictedTokens(t *testing.T) { + ctx, s, disk, convID := setupFlushTestStore(t, "flush-evicted-token") + + te := newTokenEntry() + te.MsgIDs[7] = chat1.EmptyStruct{} + s.Lock() + require.NoError(t, s.putTokenEntry(ctx, convID, "alpha", te)) + s.Unlock() + + // evicted before the flush loop got to it + s.tokenCache.Purge() + + require.NoError(t, s.Flush()) + + onDisk, err := disk.GetTokenEntry(ctx, convID, "alpha") + require.NoError(t, err) + require.NotNil(t, onDisk, "an evicted dirty token must still be written to disk") + require.Contains(t, onDisk.MsgIDs, chat1.MessageID(7)) +} + +func TestFlushWritesEvictedAliases(t *testing.T) { + ctx, s, disk, _ := setupFlushTestStore(t, "flush-evicted-alias") + + ae := newAliasEntry() + ae.Aliases["alpha"] = 1 + s.Lock() + require.NoError(t, s.putAliasEntry(ctx, "alp", ae)) + s.Unlock() + + s.aliasCache.Purge() + + require.NoError(t, s.Flush()) + + onDisk, err := disk.GetAliasEntry(ctx, "alp") + require.NoError(t, err) + require.NotNil(t, onDisk, "an evicted dirty alias must still be written to disk") + require.Contains(t, onDisk.Aliases, "alpha") +} + +// Losing metadata is the one that costs search history: SeenIDs is what records +// a message as indexed, so dropping it silently un-indexes messages whose tokens +// were already written. +func TestFlushWritesEvictedMetadata(t *testing.T) { + ctx, s, disk, convID := setupFlushTestStore(t, "flush-evicted-md") + + require.NoError(t, s.MarkSeen(ctx, convID, []chat1.MessageID{1, 2, 3})) + + s.mdCache.Purge() + + require.NoError(t, s.Flush()) + + onDisk, err := disk.GetMetadata(ctx, convID) + require.NoError(t, err) + require.NotNil(t, onDisk, "evicted dirty metadata must still be written to disk") + require.Contains(t, onDisk.SeenIDs, chat1.MessageID(2)) +} + +// An evicted dirty entry used to fall through to the copy on disk, which is by +// definition stale - it is missing precisely the writes that are still dirty. +// Further updates were then built on top of that stale copy. +func TestReadAfterEvictionDoesNotSeeStaleDisk(t *testing.T) { + ctx, s, _, convID := setupFlushTestStore(t, "flush-stale-read") + + first := newTokenEntry() + first.MsgIDs[1] = chat1.EmptyStruct{} + s.Lock() + require.NoError(t, s.putTokenEntry(ctx, convID, "alpha", first)) + s.Unlock() + require.NoError(t, s.Flush()) + + // a second write that is still dirty when the entry is evicted + second := newTokenEntry() + second.MsgIDs[1] = chat1.EmptyStruct{} + second.MsgIDs[2] = chat1.EmptyStruct{} + s.Lock() + require.NoError(t, s.putTokenEntry(ctx, convID, "alpha", second)) + s.Unlock() + + s.tokenCache.Purge() + + s.Lock() + got, err := s.getTokenEntry(ctx, convID, "alpha") + s.Unlock() + require.NoError(t, err) + require.Contains(t, got.MsgIDs, chat1.MessageID(2), + "read fell through to the stale copy on disk instead of the pending write") +} + +// Pending entries are pinned in memory until flushed, so without a size bound a +// fast writer decides how much accumulates between the 15s ticks. +func TestFlushSignalledWhenPendingSetIsFull(t *testing.T) { + ctx, s, _, convID := setupFlushTestStore(t, "flush-size-bound") + + s.Lock() + for i := 0; i < maxDirtyEntries-1; i++ { + te := newTokenEntry() + te.MsgIDs[chat1.MessageID(i)] = chat1.EmptyStruct{} + require.NoError(t, s.putTokenEntry(ctx, convID, fmt.Sprintf("tok%d", i), te)) + } + s.Unlock() + + select { + case <-s.flushNeeded(): + require.Fail(t, "asked for a flush below the bound") + default: + } + + s.Lock() + require.NoError(t, s.putTokenEntry(ctx, convID, "last", newTokenEntry())) + s.Unlock() + + select { + case <-s.flushNeeded(): + default: + require.Fail(t, "crossing the bound must ask for a flush") + } +} + +// Rewriting the same key does not grow the pending set, so it must not count +// toward the bound - otherwise a hot token flushes constantly for no reason. +func TestPendingCountTracksDistinctEntries(t *testing.T) { + ctx, s, _, convID := setupFlushTestStore(t, "flush-size-distinct") + + s.Lock() + for i := 0; i < maxDirtyEntries*2; i++ { + te := newTokenEntry() + te.MsgIDs[chat1.MessageID(i)] = chat1.EmptyStruct{} + require.NoError(t, s.putTokenEntry(ctx, convID, "same", te)) + } + count := s.dirtyCount + s.Unlock() + + require.Equal(t, 1, count, "rewrites of one key must count once") + + select { + case <-s.flushNeeded(): + require.Fail(t, "rewriting one key must not trip the bound") + default: + } +} + +func TestFlushResetsPendingCount(t *testing.T) { + ctx, s, _, convID := setupFlushTestStore(t, "flush-size-reset") + + s.Lock() + for i := 0; i < maxDirtyEntries; i++ { + require.NoError(t, s.putTokenEntry(ctx, convID, fmt.Sprintf("tok%d", i), newTokenEntry())) + } + s.Unlock() + + require.NoError(t, s.Flush()) + + s.Lock() + count := s.dirtyCount + s.Unlock() + require.Equal(t, 0, count, "flush must clear the pending count with the maps") + + // and the signal it answered must not still be queued, or the loop spins + select { + case <-s.flushNeeded(): + require.Fail(t, "a flush left its own signal pending") + default: + } +} + +func TestMetadataReadAfterEvictionKeepsSeenIDs(t *testing.T) { + ctx, s, disk, convID := setupFlushTestStore(t, "flush-stale-md-read") + + require.NoError(t, s.MarkSeen(ctx, convID, []chat1.MessageID{1})) + require.NoError(t, s.Flush()) + require.NoError(t, s.MarkSeen(ctx, convID, []chat1.MessageID{2})) + + s.mdCache.Purge() + + s.RLock() + md, err := s.getMetadataLocked(ctx, convID) + s.RUnlock() + require.NoError(t, err) + require.Contains(t, md.SeenIDs, chat1.MessageID(2), + "an unflushed SeenID was lost to a cache eviction") + + // and the id must survive the round trip rather than only living in memory + require.NoError(t, s.Flush()) + onDisk, err := disk.GetMetadata(ctx, convID) + require.NoError(t, err) + require.Contains(t, onDisk.SeenIDs, chat1.MessageID(2)) +} + +// A write that never reached disk must go back into the pending set. Before +// this, Flush cleared dirty tracking before writing, so a failed write was +// dropped: nothing marks a token entry dirty again once its message is indexed, +// while the live metadata keeps its SeenIDs. The next successful flush then +// recorded those messages as indexed with no tokens behind them. +func TestFlushRequeuesUnwrittenEntriesOnFailure(t *testing.T) { + ctx, s, disk, convID := setupFlushTestStore(t, "flush-requeue") + + te := newTokenEntry() + te.MsgIDs[7] = chat1.EmptyStruct{} + s.Lock() + require.NoError(t, s.putTokenEntry(ctx, convID, "alpha", te)) + s.Unlock() + require.NoError(t, s.MarkSeen(ctx, convID, []chat1.MessageID{7})) + + disk.Lock() + disk.failTokenPut = fmt.Errorf("disk is full") + disk.Unlock() + + require.Error(t, s.Flush(), "a failed write must surface") + + // metadata must not have landed while its token did not + onDisk, err := disk.GetMetadata(ctx, convID) + require.NoError(t, err) + require.Nil(t, onDisk, "metadata was published without the tokens behind it") + + s.Lock() + pending := s.dirtyCount + _, tokenPending := s.dirtyTokens[convID.ConvIDStr()]["alpha"] + _, mdPending := s.dirtyMetadata[convID.ConvIDStr()] + s.Unlock() + require.True(t, tokenPending, "the token that failed to write was dropped") + require.True(t, mdPending, "the metadata that was never attempted was dropped") + require.Equal(t, 2, pending, "dirtyCount must match what was put back") + + // once the disk recovers, the retry writes everything + disk.Lock() + disk.failTokenPut = nil + disk.Unlock() + require.NoError(t, s.Flush()) + + gotToken, err := disk.GetTokenEntry(ctx, convID, "alpha") + require.NoError(t, err) + require.NotNil(t, gotToken, "the retry did not write the token") + require.Contains(t, gotToken.MsgIDs, chat1.MessageID(7)) + gotMd, err := disk.GetMetadata(ctx, convID) + require.NoError(t, err) + require.NotNil(t, gotMd) + require.Contains(t, gotMd.SeenIDs, chat1.MessageID(7)) +} + +// Requeueing must not clobber a write that happened after the snapshot was +// taken - that value is newer than the one that failed. +func TestRequeueKeepsNewerPendingWrite(t *testing.T) { + ctx, s, _, convID := setupFlushTestStore(t, "flush-requeue-newer") + + stale := newTokenEntry() + stale.MsgIDs[1] = chat1.EmptyStruct{} + newer := newTokenEntry() + newer.MsgIDs[2] = chat1.EmptyStruct{} + + s.Lock() + require.NoError(t, s.putTokenEntry(ctx, convID, "alpha", newer)) + s.Unlock() + + s.requeue([]tokenSnapshot{{convID: convID, token: "alpha", entry: stale}}, nil, nil) + + s.Lock() + got := s.dirtyTokens[convID.ConvIDStr()]["alpha"] + count := s.dirtyCount + s.Unlock() + require.Contains(t, got.MsgIDs, chat1.MessageID(2), "requeue overwrote a newer pending write") + require.Equal(t, 1, count, "requeue double-counted an entry that was already pending") +} + +// Remove used to write metadata straight to disk, ahead of the tokens for any +// SeenIDs an in-flight Add had put on the same live object. +func TestRemoveRoutesMetadataThroughOverlay(t *testing.T) { + ctx, s, disk, convID := setupFlushTestStore(t, "remove-ordering") + + msg := textMsgForTest(3, "hello world") + require.NoError(t, s.Add(ctx, convID, []chat1.MessageUnboxed{msg})) + + require.NoError(t, s.Remove(ctx, convID, []chat1.MessageUnboxed{msg})) + + onDisk, err := disk.GetMetadata(ctx, convID) + require.NoError(t, err) + require.Nil(t, onDisk, + "Remove published metadata to disk directly, ahead of the pending tokens") + + s.Lock() + _, mdPending := s.dirtyMetadata[convID.ConvIDStr()] + s.Unlock() + require.True(t, mdPending, "Remove must leave metadata pending like every other writer") +} + +// Add marked a message seen before indexing it, so an indexing failure left the +// message recorded as indexed with no tokens - and nothing re-indexes a message +// the metadata already accounts for. +func TestAddDoesNotMarkSeenWhenIndexingFails(t *testing.T) { + ctx, s, disk, convID := setupFlushTestStore(t, "add-mark-after-work") + + disk.Lock() + disk.failTokenGet = fmt.Errorf("cannot read token entry") + disk.Unlock() + + msg := textMsgForTest(5, "hello world") + require.Error(t, s.Add(ctx, convID, []chat1.MessageUnboxed{msg}), + "the indexing failure must surface") + + s.RLock() + md, err := s.getMetadataLocked(ctx, convID) + s.RUnlock() + require.NoError(t, err) + require.NotContains(t, md.SeenIDs, chat1.MessageID(5), + "message was marked indexed even though indexing it failed") +} + +func textMsgForTest(id chat1.MessageID, body string) chat1.MessageUnboxed { + return chat1.NewMessageUnboxedWithValid(chat1.MessageUnboxedValid{ + ClientHeader: chat1.MessageClientHeaderVerified{ + MessageType: chat1.MessageType_TEXT, + Conv: chat1.ConversationIDTriple{TopicType: chat1.TopicType_CHAT}, + }, + MessageBody: chat1.NewMessageBodyWithText(chat1.MessageText{Body: body}), + ServerHeader: chat1.MessageServerHeader{MessageID: id}, + }) +} + +// Bumping indexMetadataVersion is how a damaged index gets rebuilt: a conv whose +// metadata claims messages are indexed reads as complete, so nothing revisits +// it and only a version change discards it. That makes the mismatch check +// load-bearing rather than incidental. +func TestStaleVersionEntriesAreDiscarded(t *testing.T) { + ctx, s, disk, convID := setupFlushTestStore(t, "stale-version") + + staleToken := newTokenEntry() + staleToken.Version = "1:1" + staleToken.MsgIDs[7] = chat1.EmptyStruct{} + require.NoError(t, disk.PutTokenEntry(ctx, convID, "alpha", staleToken)) + + staleMd := newIndexMetadata() + staleMd.Version = "1:1" + staleMd.SeenIDs[7] = chat1.EmptyStruct{} + require.NoError(t, disk.PutMetadata(ctx, convID, staleMd)) + + s.Lock() + gotToken, err := s.getTokenEntry(ctx, convID, "alpha") + s.Unlock() + require.NoError(t, err) + require.NotContains(t, gotToken.MsgIDs, chat1.MessageID(7), + "an entry written by an older index version was served instead of discarded") + + s.RLock() + gotMd, err := s.getMetadataLocked(ctx, convID) + s.RUnlock() + require.NoError(t, err) + require.NotContains(t, gotMd.SeenIDs, chat1.MessageID(7), + "stale metadata was kept, so the conv would still read as indexed") +} + +// A delete that lands while a flush is writing must survive that flush. Flush +// snapshots under the lock but writes outside it, so a delete applied straight +// to disk in that window was immediately undone by the write that followed and +// the entry stayed searchable forever. Queuing the delete instead orders it +// after the write. +func TestDeleteDuringFlushIsNotResurrected(t *testing.T) { + ctx, s, disk, convID := setupFlushTestStore(t, "delete-during-flush") + + te := newTokenEntry() + te.MsgIDs[7] = chat1.EmptyStruct{} + s.Lock() + require.NoError(t, s.putTokenEntry(ctx, convID, "alpha", te)) + s.Unlock() + + // delete the entry midway through the flush that is writing it + disk.Lock() + disk.onPutToken = func() { + s.Lock() + s.deleteTokenEntry(ctx, convID, "alpha") + s.Unlock() + } + disk.Unlock() + + require.NoError(t, s.Flush()) + // that flush wrote the value it had already snapshotted; the delete is + // pending behind it and must be applied by the next one + require.NoError(t, s.Flush()) + + onDisk, err := disk.GetTokenEntry(ctx, convID, "alpha") + require.NoError(t, err) + require.Nil(t, onDisk, + "a delete issued during a flush was undone by that flush's write") + + s.Lock() + got, err := s.getTokenEntry(ctx, convID, "alpha") + s.Unlock() + require.NoError(t, err) + require.NotContains(t, got.MsgIDs, chat1.MessageID(7), + "the resurrected entry is still readable") +} + +// A queued delete must be visible to readers immediately, rather than reading +// back the copy that is still on disk until the next flush. +func TestPendingDeleteIsNotReadFromDisk(t *testing.T) { + ctx, s, disk, convID := setupFlushTestStore(t, "pending-delete-read") + + te := newTokenEntry() + te.MsgIDs[7] = chat1.EmptyStruct{} + s.Lock() + require.NoError(t, s.putTokenEntry(ctx, convID, "alpha", te)) + s.Unlock() + require.NoError(t, s.Flush()) + + onDisk, err := disk.GetTokenEntry(ctx, convID, "alpha") + require.NoError(t, err) + require.NotNil(t, onDisk, "precondition: the entry is on disk") + + s.Lock() + s.deleteTokenEntry(ctx, convID, "alpha") + got, err := s.getTokenEntry(ctx, convID, "alpha") + s.Unlock() + require.NoError(t, err) + require.NotContains(t, got.MsgIDs, chat1.MessageID(7), + "a pending delete must not read back the copy still on disk") + + require.NoError(t, s.Flush()) + onDisk, err = disk.GetTokenEntry(ctx, convID, "alpha") + require.NoError(t, err) + require.Nil(t, onDisk, "the queued delete never reached disk") +} + +// The failure path must not undo a delete: requeue puts unwritten snapshots +// back, and a delete that happened since the snapshot has to win. +func TestRequeueDoesNotResurrectDeletedEntry(t *testing.T) { + ctx, s, _, convID := setupFlushTestStore(t, "requeue-delete") + + stale := newTokenEntry() + stale.MsgIDs[7] = chat1.EmptyStruct{} + + // the delete lands after the snapshot was taken + s.Lock() + s.deleteTokenEntry(ctx, convID, "alpha") + s.Unlock() + + s.requeue([]tokenSnapshot{{convID: convID, token: "alpha", entry: stale}}, nil, nil) + + s.Lock() + entry, ok := s.dirtyTokens[convID.ConvIDStr()]["alpha"] + s.Unlock() + require.True(t, ok, "the delete should still be pending") + require.Nil(t, entry, + "requeue resurrected an entry that was deleted after the snapshot") +} diff --git a/go/chat/search/indexer.go b/go/chat/search/indexer.go index 9cfeb78bc3fb..c84786af6626 100644 --- a/go/chat/search/indexer.go +++ b/go/chat/search/indexer.go @@ -2,6 +2,7 @@ package search import ( "context" + "errors" "sort" "sync" "time" @@ -26,14 +27,14 @@ type storageAdd struct { ctx context.Context convID chat1.ConversationID msgs []chat1.MessageUnboxed - cb chan struct{} + cb chan error } type storageRemove struct { ctx context.Context convID chat1.ConversationID msgs []chat1.MessageUnboxed - cb chan struct{} + cb chan error } type Indexer struct { @@ -62,6 +63,12 @@ type Indexer struct { selectiveSyncActive bool flushDelay time.Duration + // convID -> how a conv's last reindex attempt went. A conv whose missing + // count did not move is backed off rather than retried every interval. + // See stalledConvs in SelectiveSync. + stalledMu sync.Mutex + stalledConvs map[chat1.ConvIDStr]stalledConv + // for testing consumeCh chan chat1.ConversationID reindexCh chan chat1.ConversationID @@ -82,6 +89,7 @@ func NewIndexer(g *globals.Context) *Indexer { clock: clockwork.NewRealClock(), flushDelay: 15 * time.Second, storageCh: make(chan any, 100), + stalledConvs: make(map[chat1.ConvIDStr]stalledConv), } switch idx.G().GetAppType() { case libkb.MobileAppType: @@ -297,7 +305,20 @@ func (idx *Indexer) Stop(ctx context.Context) chan struct{} { defer idx.Unlock() ch := make(chan struct{}) if idx.started { + // Write out what is pending before dropping it. ClearMemory discards the + // overlay, and the flush loop is about to stop, so anything not written + // here is simply lost - work that has to be done again at best, and at + // worst tokens whose metadata already reached disk. + if err := idx.store.Flush(); err != nil { + idx.Debug(ctx, "Stop: final flush failed: %s", err) + } idx.store.ClearMemory() + // entries are keyed by convID only, and team convs are shared between + // users on this device: a count left by the previous user must not + // decide anything for the next one + idx.stalledMu.Lock() + idx.stalledConvs = make(map[chat1.ConvIDStr]stalledConv) + idx.stalledMu.Unlock() idx.started = false close(idx.stopCh) go func() { @@ -373,12 +394,33 @@ func (idx *Indexer) consumeResultsForTest(convID chat1.ConversationID, err error } } -func (idx *Indexer) storageDispatch(op any) { +// storageDispatch queues op, reporting whether it was accepted. A caller that +// waits on the op's callback must close it itself when this returns false: +// dropping the op silently left that callback with nobody to close it, and the +// waiter - reindexConv does an unconditional receive - parked forever. In +// SelectiveSync that leaks the sync's cancel func, so every later attempt sees +// "already running" and background indexing is done until the process restarts. +func (idx *Indexer) storageDispatch(op any) bool { select { case idx.storageCh <- op: + return true default: idx.Debug(context.Background(), "storageDispatch: failed to dispatch storage operation") + return false + } +} + +// releaseStorageCB wakes an op's waiter, reporting whether the op actually ran. +// The channel is buffered, so the send never blocks even with nobody waiting, +// and a waiter that receives from a closed channel reads nil - success. +func releaseStorageCB(cb chan error, err error) { + if cb == nil { + return + } + if err != nil { + cb <- err } + close(cb) } func (idx *Indexer) storageLoop(stopCh chan struct{}) error { @@ -388,7 +430,21 @@ func (idx *Indexer) storageLoop(stopCh chan struct{}) error { select { case <-stopCh: idx.Debug(ctx, "storageLoop: shutting down") - return nil + // Release anything still queued. These ops will not run, but a + // caller blocked on one's callback would never wake up otherwise. + for { + select { + case iop := <-idx.storageCh: + switch op := iop.(type) { + case storageAdd: + releaseStorageCB(op.cb, errStorageStopped) + case storageRemove: + releaseStorageCB(op.cb, errStorageStopped) + } + default: + return nil + } + } case iop := <-idx.storageCh: switch op := iop.(type) { case storageAdd: @@ -397,14 +453,14 @@ func (idx *Indexer) storageLoop(stopCh chan struct{}) error { idx.Debug(op.ctx, "storageLoop: add failed: %s", err) } idx.consumeResultsForTest(op.convID, err) - close(op.cb) + releaseStorageCB(op.cb, err) case storageRemove: err := idx.store.Remove(op.ctx, op.convID, op.msgs) if err != nil { idx.Debug(op.ctx, "storageLoop: remove failed: %s", err) } idx.consumeResultsForTest(op.convID, err) - close(op.cb) + releaseStorageCB(op.cb, err) } } } @@ -418,6 +474,11 @@ func (idx *Indexer) flushLoop(stopCh chan struct{}) error { case <-stopCh: idx.Debug(ctx, "flushLoop: shutting down") return nil + case <-idx.store.flushNeeded(): + // too much pending to wait out the interval + if err := idx.store.Flush(); err != nil { + idx.Debug(ctx, "flushLoop: failed to flush: %s", err) + } case <-idx.clock.After(idx.flushDelay): if err := idx.store.Flush(); err != nil { idx.Debug(ctx, "flushLoop: failed to flush: %s", err) @@ -454,8 +515,8 @@ func (idx *Indexer) Add(ctx context.Context, convID chat1.ConversationID, func (idx *Indexer) add(ctx context.Context, convID chat1.ConversationID, msgs []chat1.MessageUnboxed, force bool, -) (cb chan struct{}, err error) { - cb = make(chan struct{}) +) (cb chan error, err error) { + cb = make(chan error, 1) if idx.G().GetEnv().GetDisableSearchIndexer() { close(cb) return cb, nil @@ -472,12 +533,16 @@ func (idx *Indexer) add(ctx context.Context, convID chat1.ConversationID, defer idx.Trace(ctx, &err, "Indexer.Add conv: %v, msgs: %d, force: %v", convID, len(msgs), force)() - idx.storageDispatch(storageAdd{ + if !idx.storageDispatch(storageAdd{ ctx: globals.BackgroundChatCtx(ctx, idx.G()), convID: convID, msgs: msgs, cb: cb, - }) + }) { + // nothing will run this op, so nothing will close cb + close(cb) + return cb, errStorageQueueFull + } return cb, nil } @@ -496,8 +561,8 @@ func (idx *Indexer) Remove(ctx context.Context, convID chat1.ConversationID, func (idx *Indexer) remove(ctx context.Context, convID chat1.ConversationID, msgs []chat1.MessageUnboxed, force bool, -) (cb chan struct{}, err error) { - cb = make(chan struct{}) +) (cb chan error, err error) { + cb = make(chan error, 1) if idx.G().GetEnv().GetDisableSearchIndexer() { close(cb) return cb, nil @@ -514,12 +579,16 @@ func (idx *Indexer) remove(ctx context.Context, convID chat1.ConversationID, defer idx.Trace(ctx, &err, "Indexer.Remove conv: %v, msgs: %d, force: %v", convID, len(msgs), force)() - idx.storageDispatch(storageRemove{ + if !idx.storageDispatch(storageRemove{ ctx: globals.BackgroundChatCtx(ctx, idx.G()), convID: convID, msgs: msgs, cb: cb, - }) + }) { + // nothing will run this op, so nothing will close cb + close(cb) + return cb, errStorageQueueFull + } return cb, nil } @@ -547,67 +616,74 @@ func (idx *Indexer) reindexConv(ctx context.Context, rconv types.RemoteConversat utils.GetRemoteConvDisplayName(rconv), minIdxID, maxIdxID, len(missingIDs))() reason := chat1.GetThreadReason_INDEXED_SEARCH - if len(missingIDs) < idx.pageSize { - msgs, err := idx.G().ConvSource.GetMessages(ctx, rconv.GetConvID(), idx.uid, missingIDs, &reason, - nil, false) + // Page over the MISSING ids, not over the raw min..max range. Paging the + // range re-fetched everything already indexed (measured: 57,648 already-seen + // messages re-added in a single pass) and, because each pass restarts at the + // lowest missing id and stops once the inbox-wide budget is spent, the top of + // a large conv was never reachable at all. + for start := 0; start < len(missingIDs); start += idx.pageSize { + select { + case <-ctx.Done(): + return completedJobs, ctx.Err() + default: + } + end := start + idx.pageSize + if end > len(missingIDs) { + end = len(missingIDs) + } + chunk := missingIDs[start:end] + msgs, err := idx.G().ConvSource.GetMessages(ctx, convID, idx.uid, chunk, &reason, nil, false) if err != nil { if utils.IsPermanentErr(err) { - return 0, err + return completedJobs, err } - return 0, nil + // transient: leave these ids missing so a later pass retries them + continue } cb, err := idx.add(ctx, convID, msgs, true) if err != nil { - return 0, err + return completedJobs, err + } + select { + case addErr := <-cb: + if addErr != nil { + // The op failed or never ran, so nothing in this chunk was + // indexed. Marking it seen here would record those messages as + // indexed with no tokens behind them, and the conv would never + // be revisited to fix it. + return completedJobs, addErr + } + case <-ctx.Done(): + return completedJobs, ctx.Err() + } + // The fetch succeeded, so every id we asked for is now accounted for: + // the ones that came back were indexed by add(), and the ones that did + // not are ids the source will never produce. Without this the latter + // keep the conv permanently "not fully indexed". + if err := idx.store.MarkSeen(ctx, convID, chunk); err != nil { + // The store is failing, not the conv. Give up on this conv rather + // than paging on: without the mark these ids stay missing, so + // counting the job would make a storage failure look like a stalled + // conv and back it off for a reason that has nothing to do with it. + return completedJobs, err } - <-cb completedJobs++ - } else { - query := &chat1.GetThreadQuery{ - DisablePostProcessThread: true, - MarkAsRead: false, + if numJobs > 0 && completedJobs >= numJobs { + break } - for i := minIdxID; i < maxIdxID; i += chat1.MessageID(idx.pageSize) { //nolint:gosec // G115: pageSize is a small positive config value, safe to convert - select { - case <-ctx.Done(): - return 0, ctx.Err() - default: - } - pagination := utils.MessageIDControlToPagination(ctx, idx.DebugLabeler, &chat1.MessageIDControl{ - Num: idx.pageSize, - Pivot: &i, - Mode: chat1.MessageIDControlMode_NEWERMESSAGES, - }, nil) - tv, err := idx.G().ConvSource.Pull(ctx, convID, idx.uid, reason, nil, query, pagination) + if inboxIndexStatus != nil { + status, err := idx.store.IndexStatus(ctx, conv) if err != nil { - if utils.IsPermanentErr(err) { - return 0, err - } + idx.Debug(ctx, "updateInboxIndex: unable to get index status %v", err) continue } - cb, err := idx.add(ctx, convID, tv.Messages, true) + inboxIndexStatus.addConv(status, conv) + percentIndexed, err := inboxIndexStatus.updateUI(ctx) if err != nil { - return 0, err - } - <-cb - completedJobs++ - if numJobs > 0 && completedJobs >= numJobs { - break - } - if inboxIndexStatus != nil { - status, err := idx.store.IndexStatus(ctx, conv) - if err != nil { - idx.Debug(ctx, "updateInboxIndex: unable to get index status %v", err) - continue - } - inboxIndexStatus.addConv(status, conv) - percentIndexed, err := inboxIndexStatus.updateUI(ctx) - if err != nil { - idx.Debug(ctx, "unable to update ui %v", err) - } else { - idx.Debug(ctx, "%v is %d%% indexed, inbox is %d%% indexed", - utils.GetRemoteConvDisplayName(rconv), status.percentIndexed(), percentIndexed) - } + idx.Debug(ctx, "unable to update ui %v", err) + } else { + idx.Debug(ctx, "%v is %d%% indexed, inbox is %d%% indexed", + utils.GetRemoteConvDisplayName(rconv), status.percentIndexed(), percentIndexed) } } } @@ -728,6 +804,99 @@ func (idx *Indexer) setSelectiveSyncActive(val bool) { // SelectiveSync queues up a small number of jobs on the background loader // periodically so our index can cover all conversations. The number of jobs // varies between desktop and mobile so mobile can be more conservative. +// A stalled conv is retried after 2, then 4, then 8, then every 12 skipped +// intervals - so on desktop it settles to one attempt per 13 passes, a bit over +// an hour, reached after ~90 minutes. Even a permanently stuck conv is retried +// occasionally: nothing is suppressed forever on the strength of an inference. +// +// With reindexConv now marking fetched-but-unreturned ids as seen, a conv should +// converge on its own and never reach here. This is the backstop for whatever +// still cannot converge, not the mechanism. +// errStorageQueueFull reports that a storage op was never queued. Callers that +// mark work as done once the op completes must not do so on this error: the op +// did not run, so nothing was indexed. +var errStorageQueueFull = errors.New("search storage queue full, operation dropped") + +// errStorageStopped reports that a queued op was released by shutdown rather +// than run. Same contract as errStorageQueueFull: the op did not run. +var errStorageStopped = errors.New("search storage loop stopped, operation dropped") + +const maxStallSkips = 12 + +// the largest exponent used for the backoff; see recordIndexProgress +const maxStallStreak = 4 + +type stalledConv struct { + numMissing uint + // intervals still to skip before the next attempt + skips int + // consecutive no-progress attempts, doubling the backoff each time + streak uint +} + +// convStalled reports whether to skip this conv on this pass, and consumes one +// of its backoff intervals if so. +// +// The backoff is deliberately bounded rather than permanent. "numMissing did not +// move" is strong evidence that the missing IDs cannot be fetched, but it is not +// proof: reindexConv restarts at the lowest missing ID and stops once it has +// spent the pass budget, so a conv whose budgeted pages happened to cover only +// unfetchable IDs looks identical to a stuck one while still having real work +// further up its range. Backing off instead of suppressing means that conv +// resumes on its own, and a genuinely stuck one still costs ~1/12th of what it +// did. +func (idx *Indexer) convStalled(convID chat1.ConversationID, numMissing uint) bool { + idx.stalledMu.Lock() + defer idx.stalledMu.Unlock() + convIDStr := convID.ConvIDStr() + prev, ok := idx.stalledConvs[convIDStr] + if !ok || prev.numMissing != numMissing { + // never seen, or something changed the conv since: give it an attempt + return false + } + if prev.skips <= 0 { + return false + } + prev.skips-- + idx.stalledConvs[convIDStr] = prev + return true +} + +// recordIndexProgress stores the missing count after an attempt, so the next +// pass can tell whether that attempt accomplished anything. A conv that has +// reached 0 needs no entry: FullyIndexed already skips it. +func (idx *Indexer) recordIndexProgress(ctx context.Context, conv chat1.Conversation, before uint) { + status, err := idx.store.IndexStatus(ctx, conv) + if err != nil { + idx.Debug(ctx, "recordIndexProgress: unable to get index status: %v", err) + return + } + convIDStr := conv.GetConvID().ConvIDStr() + idx.stalledMu.Lock() + defer idx.stalledMu.Unlock() + if status.fullyIndexed() || status.numMissing != before { + // done, or made progress: clear any marker so it keeps getting attempts + delete(idx.stalledConvs, convIDStr) + return + } + // Clamp the EXPONENT, not the result: streak is unbounded, and a large + // enough shift overflows int negative (then to 0), which reads as "no skips" + // and silently turns the backoff off for good. + streak := idx.stalledConvs[convIDStr].streak + 1 + if streak > maxStallStreak { + streak = maxStallStreak + } + skips := 1 << streak + if skips > maxStallSkips { + skips = maxStallSkips + } + idx.stalledConvs[convIDStr] = stalledConv{ + numMissing: status.numMissing, + skips: skips, + streak: streak, + } +} + func (idx *Indexer) SelectiveSync(ctx context.Context) (err error) { defer idx.Trace(ctx, &err, "SelectiveSync")() defer idx.PerfTrace(ctx, &err, "SelectiveSync")() @@ -750,22 +919,52 @@ func (idx *Indexer) SelectiveSync(ctx context.Context) (err error) { default: } convID := conv.GetConvID() - fullyIndexed, err := idx.store.FullyIndexed(ctx, conv.Conv) + // one status read, not one per question: each is an O(maxID-minID) scan + // of SeenIDs under the store lock + status, err := idx.store.IndexStatus(ctx, conv.Conv) if err != nil { idx.Debug(ctx, "SelectiveSync: Unable to get md for conv: %v, %v", convID, err) continue } - if fullyIndexed { + if status.fullyIndexed() { + continue + } + + // A conv is "fully indexed" only when numMissing hits 0, and numMissing + // counts every ID in the min..max range. IDs the server will never + // return - deleted, or never delivered to us - therefore stay missing + // forever, so such a conv is never fully indexed and SelectiveSync + // re-queues it every syncInterval. Each pass re-fetches thousands of + // messages that are already indexed, marks nothing new, and burns the + // whole maxSyncConvs budget: measured at ~1800 remote getMessages every + // 5 minutes, indefinitely, on an idle desktop client, with numMissing + // byte-identical across 7 consecutive cycles. + // + // So skip a conv whose missing count did not move on its last attempt. + // Anything that actually changes the conv - a new message, a delete, + // clearing the index - changes the count and lets it back in, so this + // only suppresses the provably unproductive case. + if idx.convStalled(convID, status.numMissing) { continue } completedJobs, err := idx.reindexConv(ctx, conv, numJobs, nil) if err != nil { idx.Debug(ctx, "Unable to reindex conv: %v, %v", convID, err) + // the pages it did finish still spent budget + numJobs -= completedJobs + if numJobs <= 0 { + break + } continue - } else if completedJobs == 0 { + } + if completedJobs == 0 { + // also the transient-fetch-failure path in reindexConv, which is + // indistinguishable from no progress here. Don't record a stall for + // it: a network blip would suppress the conv until it next changes. continue } + idx.recordIndexProgress(ctx, conv.Conv, status.numMissing) idx.Debug(ctx, "SelectiveSync: Indexed completed jobs %d", completedJobs) numJobs -= completedJobs if numJobs <= 0 { @@ -861,6 +1060,11 @@ func (idx *Indexer) Clear(ctx context.Context, uid gregor1.UID, convID chat1.Con if store == nil { return nil } + // clearing the index resets what is missing, so this conv gets fresh + // attempts rather than staying suppressed at its old count + idx.stalledMu.Lock() + delete(idx.stalledConvs, convID.ConvIDStr()) + idx.stalledMu.Unlock() return store.Clear(ctx, uid, convID) } @@ -871,6 +1075,9 @@ func (idx *Indexer) OnDbNuke(mctx libkb.MetaContext) (err error) { if !idx.started { return nil } + idx.stalledMu.Lock() + idx.stalledConvs = make(map[chat1.ConvIDStr]stalledConv) + idx.stalledMu.Unlock() idx.store.ClearMemory() return nil } diff --git a/go/chat/search/md.go b/go/chat/search/md.go index e053befd3711..0f0d4bcbc60c 100644 --- a/go/chat/search/md.go +++ b/go/chat/search/md.go @@ -9,7 +9,22 @@ import ( "github.com/keybase/client/go/protocol/chat1" ) -const indexMetadataVersion = 3 +// 3 -> 4 discards every conversation's index metadata, which is what makes a +// damaged index repair itself. Several bugs could leave the metadata recording +// messages as indexed while the tokens making them findable never reached disk: +// writes evicted from the cache before a flush, a flush that failed partway, +// Remove publishing metadata ahead of its tokens, and Add marking a message +// before indexing it. Such a conversation reads as fully indexed, so nothing +// revisits it and the messages stay unsearchable. +// +// Dropping the metadata alone - rather than bumping indexVersion, which would +// discard the token and alias entries too - means every message is re-indexed +// onto the entries already on disk, so search keeps working throughout the +// repair instead of going dark until the rebuild catches up. The cost is that +// surviving alias refcounts are inflated further by the re-add, and that token +// entries naming since-deleted messages survive; those are filtered at search +// time by the message lookup, so they cost a lookup, not a wrong result. +const indexMetadataVersion = 4 type indexMetadata struct { SeenIDs map[chat1.MessageID]chat1.EmptyStruct `codec:"s"` diff --git a/go/chat/search/stall_test.go b/go/chat/search/stall_test.go new file mode 100644 index 000000000000..90a487750a4c --- /dev/null +++ b/go/chat/search/stall_test.go @@ -0,0 +1,241 @@ +package search + +import ( + "context" + "fmt" + "testing" + "time" + + "github.com/keybase/client/go/chat/globals" + "github.com/keybase/client/go/chat/types" + "github.com/keybase/client/go/externalstest" + "github.com/keybase/client/go/protocol/chat1" + "github.com/keybase/client/go/protocol/gregor1" + "github.com/stretchr/testify/require" +) + +// The stall backoff only fires for a conv that cannot converge, which a healthy +// client never produces: SelectiveSync drains every conv to numMissing 0 and +// skips it thereafter. These paths are therefore unreachable from an end-to-end +// run and can only be covered here. + +// add() converts its context with globals.BackgroundChatCtx, which reaches for +// the context factory. Nothing on the paths under test uses what it returns, so +// a stub keeps these tests from needing the whole chat harness. +type stubCtxFactory struct{} + +func (stubCtxFactory) NewKeyFinder() types.KeyFinder { return nil } +func (stubCtxFactory) NewUPAKFinder() types.UPAKFinder { return nil } + +func setupStallTestIndexer(t *testing.T, label string) (*Indexer, chat1.Conversation) { + tc := externalstest.SetupTest(t, label, 2) + t.Cleanup(tc.Cleanup) + g := globals.NewContext(tc.G, &globals.ChatContext{CtxFactory: stubCtxFactory{}}) + idx := NewIndexer(g) + idx.SetUID(gregor1.UID([]byte{1, 2, 3, 4})) + + convID := chat1.ConversationID([]byte{ + 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, + }) + conv := chat1.Conversation{ + Metadata: chat1.ConversationMetadata{ConversationID: convID}, + // nothing is indexed, so this conv reads as 100 missing ids + MaxMsgSummaries: []chat1.MessageSummary{{MsgID: 100}}, + } + return idx, conv +} + +// A failed store.Add was logged and then reported to the caller as a completed +// op, so reindexConv marked the whole chunk seen against messages that never +// made it into the index. +func TestStorageLoopReportsAddFailure(t *testing.T) { + idx, _ := setupStallTestIndexer(t, "storage-add-failure") + disk := newMemDiskStorage() + idx.store.diskStorage = disk + disk.Lock() + disk.failTokenGet = fmt.Errorf("cannot read token entry") + disk.Unlock() + + stopCh := make(chan struct{}) + go func() { _ = idx.storageLoop(stopCh) }() + t.Cleanup(func() { close(stopCh) }) + + convID := chat1.ConversationID([]byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16}) + cb := make(chan error, 1) + idx.storageCh <- storageAdd{ + ctx: context.TODO(), + convID: convID, + msgs: []chat1.MessageUnboxed{textMsgForTest(1, "hello world")}, + cb: cb, + } + + select { + case err := <-cb: + require.Error(t, err, + "a failed add reported success, so its messages get marked indexed") + case <-time.After(5 * time.Second): + require.Fail(t, "the caller was never woken") + } +} + +func TestConvStalledBackoff(t *testing.T) { + ctx := context.TODO() + idx, conv := setupStallTestIndexer(t, "stall-backoff") + convID := conv.GetConvID() + + // a conv never seen before is always given an attempt + require.False(t, idx.convStalled(convID, 100), + "a conv with no recorded history must not be skipped") + + // an attempt that moved nothing earns a backoff + idx.recordIndexProgress(ctx, conv, 100) + require.True(t, idx.convStalled(convID, 100), "first skip") + require.True(t, idx.convStalled(convID, 100), "second skip") + require.False(t, idx.convStalled(convID, 100), + "backoff must expire rather than suppress the conv permanently") + + // a different missing count means something changed the conv, so it gets an + // attempt immediately whatever the backoff said + idx.recordIndexProgress(ctx, conv, 100) + require.False(t, idx.convStalled(convID, 99), + "a changed missing count must clear the backoff") +} + +func TestConvStalledClearedByProgress(t *testing.T) { + ctx := context.TODO() + idx, conv := setupStallTestIndexer(t, "stall-progress") + convID := conv.GetConvID() + + idx.recordIndexProgress(ctx, conv, 100) + require.True(t, idx.convStalled(convID, 100)) + + // "before" higher than the current count is an attempt that made progress: + // the marker must be dropped so the conv keeps getting full attempts + idx.recordIndexProgress(ctx, conv, 500) + require.False(t, idx.convStalled(convID, 100), + "a conv that made progress must not stay backed off") + + idx.stalledMu.Lock() + _, ok := idx.stalledConvs[convID.ConvIDStr()] + idx.stalledMu.Unlock() + require.False(t, ok, "progress must delete the entry, not just zero it") +} + +// TestStallStreakDoesNotOverflow pins the clamp on the shift exponent. streak is +// unbounded, so `1 << streak` on a conv that never converges eventually shifts +// past the width of an int: the result goes negative and then to 0, which reads +// as "no skips left" and silently disables the backoff for good. Clamping the +// result instead of the exponent does not help - the shift has already +// overflowed by then. +func TestStallStreakDoesNotOverflow(t *testing.T) { + ctx := context.TODO() + idx, conv := setupStallTestIndexer(t, "stall-overflow") + convID := conv.GetConvID() + + // well past the 64 rounds it takes to shift an int to zero + for i := 0; i < 200; i++ { + idx.recordIndexProgress(ctx, conv, 100) + + idx.stalledMu.Lock() + entry := idx.stalledConvs[convID.ConvIDStr()] + idx.stalledMu.Unlock() + + require.Greater(t, entry.skips, 0, + "round %d: skips fell to %d, so the backoff is off", i, entry.skips) + require.LessOrEqual(t, entry.skips, maxStallSkips, + "round %d: skips %d exceeds the cap", i, entry.skips) + require.LessOrEqual(t, entry.streak, uint(maxStallStreak), + "round %d: streak %d was not clamped", i, entry.streak) + + // consume the backoff so the next round records another no-progress + // attempt rather than sitting on this one + for idx.convStalled(convID, 100) { + } + } +} + +func TestStopResetsStalledConvs(t *testing.T) { + ctx := context.TODO() + idx, conv := setupStallTestIndexer(t, "stall-stop") + convID := conv.GetConvID() + + idx.recordIndexProgress(ctx, conv, 100) + require.True(t, idx.convStalled(convID, 100), "conv is backed off before Stop") + + idx.Lock() + idx.started = true + idx.stopCh = make(chan struct{}) + idx.Unlock() + <-idx.Stop(ctx) + + // entries are keyed by convID alone and team convs are shared between users + // on one device, so a count left by the previous user must not decide + // anything for the next one + require.False(t, idx.convStalled(convID, 100), + "Stop must clear the backoff so the next user starts clean") +} + +func TestClearResetsStalledConv(t *testing.T) { + ctx := context.TODO() + idx, conv := setupStallTestIndexer(t, "stall-clear") + convID := conv.GetConvID() + + idx.recordIndexProgress(ctx, conv, 100) + require.True(t, idx.convStalled(convID, 100), "conv is backed off before Clear") + + require.NoError(t, idx.Clear(ctx, idx.uid, convID)) + + // clearing the index resets what is missing, so the old count must not keep + // the conv suppressed + require.False(t, idx.convStalled(convID, 100), + "Clear must drop the backoff for that conv") +} + +// A dropped storage op used to leave its callback unclosed, and reindexConv +// waits on that callback. The waiter parked forever, which leaked SelectiveSync's +// cancel func and left every later pass reporting "already running" - background +// indexing dead until the process restarted. +func TestDroppedStorageOpDoesNotStrandCaller(t *testing.T) { + ctx := context.TODO() + idx, conv := setupStallTestIndexer(t, "dispatch-full") + convID := conv.GetConvID() + + // fill the dispatch queue so the next op cannot be accepted + for len(idx.storageCh) < cap(idx.storageCh) { + idx.storageCh <- storageAdd{} + } + + msgs := []chat1.MessageUnboxed{textMsgForTest(1, "hello world")} + cb, err := idx.add(ctx, convID, msgs, true) + require.ErrorIs(t, err, errStorageQueueFull, + "a dropped op must be reported, or the caller marks unindexed messages as seen") + + select { + case <-cb: + case <-time.After(5 * time.Second): + require.Fail(t, "callback of a dropped op was never closed; caller would hang forever") + } +} + +// Ops still queued when the loop shuts down have nobody left to run them, so +// their callbacks have to be released - and released with an error. Waking the +// caller with a bare "done" is worse than hanging it: reindexConv then marks the +// chunk seen, recording messages as indexed that were never handed to the store, +// and a conv that reaches numMissing 0 that way is never revisited. +func TestStorageLoopReleasesQueuedCallbacksOnShutdown(t *testing.T) { + idx, _ := setupStallTestIndexer(t, "dispatch-shutdown") + stopCh := make(chan struct{}) + cb := make(chan error, 1) + idx.storageCh <- storageAdd{cb: cb} + + close(stopCh) + require.NoError(t, idx.storageLoop(stopCh)) + + select { + case err := <-cb: + require.ErrorIs(t, err, errStorageStopped, + "an op that never ran reported success, so its messages get marked indexed") + case <-time.After(5 * time.Second): + require.Fail(t, "a queued op's callback was abandoned at shutdown") + } +} diff --git a/go/chat/search/storage.go b/go/chat/search/storage.go index d8bbd90b7aae..d2f6ffc4b7a3 100644 --- a/go/chat/search/storage.go +++ b/go/chat/search/storage.go @@ -29,6 +29,31 @@ const ( aliasDiskVersion = 1 ) +// Pending entries are held until the next flush, so a long flush interval and a +// fast writer together decide how much lives in memory. An index pass writes +// thousands of tokens a second, which would otherwise sit there for the whole +// flushDelay. This bounds the pending set instead of the clock: cross it and a +// flush is asked for immediately. +// +// It is a trigger, not a ceiling. The flush cannot run until the writer releases +// s.Lock, and one Add writes a whole batch of tokens and aliases under a single +// hold, so the set peaks at roughly this plus one batch. +// +// It therefore has to sit well above one batch. A page of 300 messages produces +// ~2600 entries, so a 1000 bound was crossed by every single batch and turned +// into "flush after every batch": measured at 65 flushes in 2.9s, a 0.04s median +// gap, rewriting hot tokens to disk - each one re-encrypted whole, not as a +// delta - on every pass instead of once an interval. Raising it to 20000 cut +// that to 7 flushes over a comparable window. +// +// Note this counts entries, not bytes, and the two differ by a lot: an entry +// holding one message id and one holding ten thousand both count as 1. A fresh +// index is a few MB at this bound, but a mature conversation's entries are much +// larger - the pending set is biased towards hot tokens, whose posting lists are +// the longest - so 35-40MB is the realistic ceiling there. A byte-based bound +// would measure the thing that actually matters. +const maxDirtyEntries = 20000 + type tokenEntry struct { Version string `codec:"v"` MsgIDs map[chat1.MessageID]chat1.EmptyStruct `codec:"m"` @@ -256,10 +281,21 @@ type store struct { mdCache *lru.Cache diskStorage diskStorage - // Track dirty entries that need to be flushed to disk - dirtyTokens map[chat1.ConvIDStr]map[string]struct{} // map[convIDStr][token] - dirtyAliases map[string]struct{} // map[alias] - dirtyMetadata map[chat1.ConvIDStr]struct{} // map[convIDStr] + // Entries written but not yet on disk. These hold the entry itself, not just + // its key: the caches above are bounded LRUs, so a dirty entry can be evicted + // before the next flush. Keying alone meant such an entry was skipped by + // Flush and lost, while a subsequent read fell through to the stale copy on + // disk and built further updates on top of it. Reads therefore consult these + // before disk, and Flush writes from these rather than from the caches. + dirtyTokens map[chat1.ConvIDStr]map[string]*tokenEntry // map[convIDStr][token] + dirtyAliases map[string]*aliasEntry // map[alias] + dirtyMetadata map[chat1.ConvIDStr]*indexMetadata // map[convIDStr] + + // how many entries are pending across all three maps above, and a one-slot + // nudge to the flush loop for when that gets big. Buffered and sent to + // without blocking: a signal already waiting is the same request. + dirtyCount int + flushNeededCh chan struct{} flushMtx sync.Mutex // Synchronizes flush operations to disk } @@ -283,9 +319,10 @@ func newStore(g *globals.Context, uid gregor1.UID) *store { tokenCache: tc, mdCache: mc, diskStorage: newDiskStore(g, uid, keyFn, encrypteddb.New(g.ExternalG(), dbFn, keyFn), g.LocalChatDb), - dirtyTokens: make(map[chat1.ConvIDStr]map[string]struct{}), - dirtyAliases: make(map[string]struct{}), - dirtyMetadata: make(map[chat1.ConvIDStr]struct{}), + dirtyTokens: make(map[chat1.ConvIDStr]map[string]*tokenEntry), + dirtyAliases: make(map[string]*aliasEntry), + dirtyMetadata: make(map[chat1.ConvIDStr]*indexMetadata), + flushNeededCh: make(chan struct{}, 1), } } @@ -471,6 +508,15 @@ func (s *store) getTokenEntry(ctx context.Context, convID chat1.ConversationID, if te, ok := s.tokenCache.Get(cacheKey); ok { return te.(*tokenEntry), nil } + // evicted from the cache but not yet flushed: the copy on disk is stale + if te, ok := s.dirtyTokens[convID.ConvIDStr()][token]; ok { + if te == nil { + // pending delete: the copy still on disk is about to go + return newTokenEntry(), nil + } + s.tokenCache.Add(cacheKey, te) + return te, nil + } if res, err = s.diskStorage.GetTokenEntry(ctx, convID, token); err != nil { return nil, err } @@ -489,6 +535,15 @@ func (s *store) getAliasEntry(ctx context.Context, alias string) (res *aliasEntr if dat, ok := s.aliasCache.Get(alias); ok { return dat.(*aliasEntry), nil } + // evicted from the cache but not yet flushed: the copy on disk is stale + if ae, ok := s.dirtyAliases[alias]; ok { + if ae == nil { + // pending delete: the copy still on disk is about to go + return newAliasEntry(), nil + } + s.aliasCache.Add(alias, ae) + return ae, nil + } if res, err = s.diskStorage.GetAliasEntry(ctx, alias); err != nil { return nil, err } @@ -511,16 +566,42 @@ func (s *store) putTokenEntry(ctx context.Context, convID chat1.ConversationID, convIDStr := convID.ConvIDStr() if s.dirtyTokens[convIDStr] == nil { - s.dirtyTokens[convIDStr] = make(map[string]struct{}) + s.dirtyTokens[convIDStr] = make(map[string]*tokenEntry) + } + if _, ok := s.dirtyTokens[convIDStr][token]; !ok { + s.dirtyCount++ } - s.dirtyTokens[convIDStr][token] = struct{}{} + s.dirtyTokens[convIDStr][token] = te + s.signalFlushIfFullLocked() return nil } +// signalFlushIfFullLocked asks the flush loop to run early once enough entries +// are pending. Callers hold s.Lock; the send never blocks, so it cannot deadlock +// against the flush it is asking for. +func (s *store) signalFlushIfFullLocked() { + if s.dirtyCount < maxDirtyEntries { + return + } + select { + case s.flushNeededCh <- struct{}{}: + default: + } +} + +// flushNeeded fires when the pending set has grown past maxDirtyEntries. +func (s *store) flushNeeded() <-chan struct{} { + return s.flushNeededCh +} + func (s *store) putAliasEntry(ctx context.Context, alias string, ae *aliasEntry) (err error) { s.aliasCache.Add(alias, ae) - s.dirtyAliases[alias] = struct{}{} + if _, ok := s.dirtyAliases[alias]; !ok { + s.dirtyCount++ + } + s.dirtyAliases[alias] = ae + s.signalFlushIfFullLocked() return nil } @@ -528,7 +609,11 @@ func (s *store) putAliasEntry(ctx context.Context, alias string, ae *aliasEntry) func (s *store) putMetadata(ctx context.Context, convID chat1.ConversationID, md *indexMetadata) (err error) { convIDStr := convID.ConvIDStr() s.mdCache.Add(convIDStr, md) - s.dirtyMetadata[convIDStr] = struct{}{} + if _, ok := s.dirtyMetadata[convIDStr]; !ok { + s.dirtyCount++ + } + s.dirtyMetadata[convIDStr] = md + s.signalFlushIfFullLocked() return nil } @@ -540,24 +625,30 @@ func (s *store) deleteTokenEntry(ctx context.Context, convID chat1.ConversationI s.tokenCache.Remove(cacheKey) + // Queue the delete rather than writing it through. Flush snapshots under the + // lock but writes outside it, so a delete applied straight to disk in that + // window was undone by the write that followed, and requeueing a failed + // write could put a deleted entry back. Ordering both through the pending + // set keeps the last operation the one that lands. convIDStr := convID.ConvIDStr() - if tokens, ok := s.dirtyTokens[convIDStr]; ok { - delete(tokens, token) - // Clean up empty map - if len(tokens) == 0 { - delete(s.dirtyTokens, convIDStr) - } + if s.dirtyTokens[convIDStr] == nil { + s.dirtyTokens[convIDStr] = make(map[string]*tokenEntry) } - - // Delete from disk immediately - s.diskStorage.RemoveTokenEntry(ctx, convID, token) + if _, pending := s.dirtyTokens[convIDStr][token]; !pending { + s.dirtyCount++ + } + s.dirtyTokens[convIDStr][token] = nil + s.signalFlushIfFullLocked() } func (s *store) deleteAliasEntry(ctx context.Context, alias string) { s.aliasCache.Remove(alias) - delete(s.dirtyAliases, alias) - // Delete from disk immediately - s.diskStorage.RemoveAliasEntry(ctx, alias) + // queued, not written through - see deleteTokenEntry + if _, pending := s.dirtyAliases[alias]; !pending { + s.dirtyCount++ + } + s.dirtyAliases[alias] = nil + s.signalFlushIfFullLocked() } // addTokens add the given tokens to the index under the given message @@ -716,6 +807,39 @@ func (s *store) ConvIndexStats(ctx context.Context, conv chat1.Conversation) (re }, nil } +// MarkSeen records IDs as accounted for without indexing anything for them. +// +// A conv is "fully indexed" only when every ID between its min and max is in +// SeenIDs, but an ID the server will not return for us can never get there by +// being indexed: deleted messages, and gaps that never existed. Before this, +// those IDs kept numMissing above zero forever, so the conv was never fully +// indexed and SelectiveSync re-fetched the same already-indexed messages every +// interval, permanently. Callers mark the IDs they asked for after a fetch that +// succeeded - the source affirmatively answered for that range, so anything +// absent from the reply is not coming. +func (s *store) MarkSeen(ctx context.Context, convID chat1.ConversationID, ids []chat1.MessageID) (err error) { + if len(ids) == 0 { + return nil + } + s.Lock() + defer s.Unlock() + md, err := s.getMetadataLocked(ctx, convID) + if err != nil { + return err + } + modified := false + for _, id := range ids { + if _, ok := md.SeenIDs[id]; !ok { + md.SeenIDs[id] = chat1.EmptyStruct{} + modified = true + } + } + if !modified { + return nil + } + return s.putMetadata(ctx, convID, md) +} + // getMetadataLocked returns the live cached metadata for convID, populating the // cache from disk on a miss. The returned *indexMetadata is shared and its // SeenIDs map may be mutated, so callers must hold s.RLock for read-only access @@ -725,6 +849,12 @@ func (s *store) getMetadataLocked(ctx context.Context, convID chat1.Conversation if cached, ok := s.mdCache.Get(convIDStr); ok { return cached.(*indexMetadata), nil } + // evicted from the cache but not yet flushed: the copy on disk is stale, and + // for metadata that means losing SeenIDs and re-indexing what it recorded + if md, ok := s.dirtyMetadata[convIDStr]; ok { + s.mdCache.Add(convIDStr, md) + return md, nil + } if res, err = s.diskStorage.GetMetadata(ctx, convID); err != nil { return nil, err @@ -758,6 +888,14 @@ func (s *store) Add(ctx context.Context, convID chat1.ConversationID, } reason := chat1.GetThreadReason_INDEXED_SEARCH superseded := make(map[chat1.MessageID]supersededFetch, len(msgs)) + + // Collect what every message in this batch supersedes, then fetch the whole + // set in one call. Asking per message meant one GetMessages round trip per + // edit or attachment upload: a backfill of a large conversation issued tens + // of thousands of single-message fetches. + superIDsByMsg := make(map[chat1.MessageID][]chat1.MessageID, len(msgs)) + var allSuperIDs []chat1.MessageID + seenSuperID := make(map[chat1.MessageID]bool) for _, msg := range msgs { switch msg.GetMessageType() { case chat1.MessageType_ATTACHMENTUPLOADED, chat1.MessageType_EDIT: @@ -766,20 +904,62 @@ func (s *store) Add(ctx context.Context, convID chat1.ConversationID, s.Debug(ctx, "Add: unable to get supersedes: %v", err) continue } - supersededMsgs, err := s.G().ChatHelper.GetMessages(ctx, s.uid, convID, superIDs, - false /* resolveSupersedes */, &reason) - if err != nil { - s.Debug(ctx, "Add: unable to fetch superseded messages: %v", err) - continue + superIDsByMsg[msg.GetMessageID()] = superIDs + for _, superID := range superIDs { + if !seenSuperID[superID] { + seenSuperID[superID] = true + allSuperIDs = append(allSuperIDs, superID) + } + } + } + } + + supersededByID := make(map[chat1.MessageID]chat1.MessageUnboxed, len(allSuperIDs)) + if len(allSuperIDs) > 0 { + supersededMsgs, err := s.G().ChatHelper.GetMessages(ctx, s.uid, convID, allSuperIDs, + false /* resolveSupersedes */, &reason) + if err != nil { + // the batch tells us nothing about which ID was at fault, so fall back + // to the per-message fetches and let each one fail on its own + s.Debug(ctx, "Add: unable to fetch superseded messages in bulk: %v", err) + for _, superIDs := range superIDsByMsg { + single, err := s.G().ChatHelper.GetMessages(ctx, s.uid, convID, superIDs, + false /* resolveSupersedes */, &reason) + if err != nil { + s.Debug(ctx, "Add: unable to fetch superseded messages: %v", err) + continue + } + for _, sm := range single { + supersededByID[sm.GetMessageID()] = sm + } } - fetch := supersededFetch{msgs: supersededMsgs} - if msg.GetMessageType() == chat1.MessageType_EDIT { - fetch.tokens = tokensFromMsg(msg) + } else { + for _, sm := range supersededMsgs { + supersededByID[sm.GetMessageID()] = sm } - superseded[msg.GetMessageID()] = fetch } } + for _, msg := range msgs { + superIDs, ok := superIDsByMsg[msg.GetMessageID()] + if !ok { + continue + } + fetch := supersededFetch{} + for _, superID := range superIDs { + if sm, ok := supersededByID[superID]; ok { + fetch.msgs = append(fetch.msgs, sm) + } + } + if len(fetch.msgs) == 0 { + continue + } + if msg.GetMessageType() == chat1.MessageType_EDIT { + fetch.tokens = tokensFromMsg(msg) + } + superseded[msg.GetMessageID()] = fetch + } + s.Lock() defer s.Unlock() @@ -800,26 +980,29 @@ func (s *store) Add(ctx context.Context, convID chat1.ConversationID, if _, ok := seenIDs[msg.GetMessageID()]; ok { continue } - modified = true - seenIDs[msg.GetMessageID()] = chat1.EmptyStruct{} + // Mark seen only once the indexing behind the mark has succeeded. md is + // the live shared object and is usually already pending, so marking up + // front committed the mark whatever happened next: an error here left + // the message recorded as indexed with no tokens for it, and nothing + // re-indexes a message the metadata already accounts for. // NOTE DELETE and DELETEHISTORY are handled through calls to `remove`, // other messages will be added if there is any content that can be // indexed. switch msg.GetMessageType() { case chat1.MessageType_ATTACHMENTUPLOADED: for _, sm := range superseded[msg.GetMessageID()].msgs { - seenIDs[sm.GetMessageID()] = chat1.EmptyStruct{} err := s.addMsg(ctx, convID, sm) if err != nil { return err } + seenIDs[sm.GetMessageID()] = chat1.EmptyStruct{} + modified = true } case chat1.MessageType_EDIT: fetch := superseded[msg.GetMessageID()] // remove the original message text and replace it with the edited // contents (using the original id in the index) for _, sm := range fetch.msgs { - seenIDs[sm.GetMessageID()] = chat1.EmptyStruct{} err := s.removeMsg(ctx, convID, sm) if err != nil { return err @@ -828,6 +1011,8 @@ func (s *store) Add(ctx context.Context, convID chat1.ConversationID, if err != nil { return err } + seenIDs[sm.GetMessageID()] = chat1.EmptyStruct{} + modified = true } default: err := s.addMsg(ctx, convID, msg) @@ -835,6 +1020,8 @@ func (s *store) Add(ctx context.Context, convID chat1.ConversationID, return err } } + seenIDs[msg.GetMessageID()] = chat1.EmptyStruct{} + modified = true } return nil } @@ -860,14 +1047,18 @@ func (s *store) Remove(ctx context.Context, convID chat1.ConversationID, continue } modified = true - seenIDs[msg.GetMessageID()] = chat1.EmptyStruct{} err := s.removeMsg(ctx, convID, msg) if err != nil { return err } } if modified { - return s.diskStorage.PutMetadata(ctx, convID, md) + // Through the overlay, never straight to disk. md is the live shared + // object, so it carries SeenIDs from an Add whose token entries are + // still only pending; writing it here published "these messages are + // indexed" ahead of the tokens backing them, and a conv that reaches + // numMissing 0 that way is never looked at again. + return s.putMetadata(ctx, convID, md) } return nil } @@ -881,16 +1072,33 @@ func (s *store) ClearMemory() { s.tokenCache.Purge() s.mdCache.Purge() - s.dirtyTokens = make(map[chat1.ConvIDStr]map[string]struct{}) - s.dirtyAliases = make(map[string]struct{}) - s.dirtyMetadata = make(map[chat1.ConvIDStr]struct{}) + s.dirtyTokens = make(map[chat1.ConvIDStr]map[string]*tokenEntry) + s.dirtyAliases = make(map[string]*aliasEntry) + s.dirtyMetadata = make(map[chat1.ConvIDStr]*indexMetadata) + s.dirtyCount = 0 + select { + case <-s.flushNeededCh: + default: + } } func (s *store) Clear(ctx context.Context, uid gregor1.UID, convID chat1.ConversationID) error { + // ClearMemory is global while the disk clear is for one conv, so clearing a + // single conversation used to throw away every other conversation's pending + // writes. Flush first so only this conv's index is actually lost. + if err := s.Flush(); err != nil { + s.Debug(ctx, "Clear: flush before clear failed: %s", err) + } s.ClearMemory() return s.diskStorage.Clear(ctx, uid, convID) } +type tokenSnapshot struct { + convID chat1.ConversationID + token string + entry *tokenEntry +} + func (s *store) Flush() error { ctx := context.Background() defer s.Trace(ctx, nil, "store.Flush")() @@ -898,12 +1106,7 @@ func (s *store) Flush() error { s.flushMtx.Lock() defer s.flushMtx.Unlock() - // Snapshot the cache entries that need to be flushed to disk. - type tokenSnapshot struct { - convID chat1.ConversationID - token string - entry *tokenEntry - } + // Snapshot the entries that need to be flushed to disk. var tokenSnapshots []tokenSnapshot aliasSnapshots := make(map[string]*aliasEntry) mdSnapshots := make(map[chat1.ConvIDStr]*indexMetadata) @@ -922,37 +1125,33 @@ func (s *store) Flush() error { s.Debug(ctx, "Flush: invalid convID %s: %s", convIDStr, err) continue } - for token := range tokens { - cacheKey := s.tokenCacheKey(convID, token) - if cachedVal, ok := s.tokenCache.Get(cacheKey); ok { - te := cachedVal.(*tokenEntry) - tokenSnapshots = append(tokenSnapshots, tokenSnapshot{ - convID: convID, - token: token, - entry: te.dup(), - }) - } + for token, te := range tokens { + tokenSnapshots = append(tokenSnapshots, tokenSnapshot{ + convID: convID, + token: token, + entry: te.dup(), + }) } } - for alias := range s.dirtyAliases { - if cachedVal, ok := s.aliasCache.Get(alias); ok { - ae := cachedVal.(*aliasEntry) - aliasSnapshots[alias] = ae.dup() - } + for alias, ae := range s.dirtyAliases { + aliasSnapshots[alias] = ae.dup() } - for convIDStr := range s.dirtyMetadata { - if cachedVal, ok := s.mdCache.Get(convIDStr); ok { - md := cachedVal.(*indexMetadata) - mdSnapshots[convIDStr] = md.dup() - } + for convIDStr, md := range s.dirtyMetadata { + mdSnapshots[convIDStr] = md.dup() } // Clear dirty tracking - s.dirtyTokens = make(map[chat1.ConvIDStr]map[string]struct{}) - s.dirtyAliases = make(map[string]struct{}) - s.dirtyMetadata = make(map[chat1.ConvIDStr]struct{}) + s.dirtyTokens = make(map[chat1.ConvIDStr]map[string]*tokenEntry) + s.dirtyAliases = make(map[string]*aliasEntry) + s.dirtyMetadata = make(map[chat1.ConvIDStr]*indexMetadata) + s.dirtyCount = 0 + // drop a signal raised before this flush: it has just been answered + select { + case <-s.flushNeededCh: + default: + } s.Unlock() } @@ -960,31 +1159,92 @@ func (s *store) Flush() error { s.Debug(ctx, "Flush: writing %d tokens, %d aliases, %d metadata to disk", len(tokenSnapshots), len(aliasSnapshots), len(mdSnapshots)) - for _, snapshot := range tokenSnapshots { + for i, snapshot := range tokenSnapshots { + // a nil entry is a queued delete, not a value to write + if snapshot.entry == nil { + s.diskStorage.RemoveTokenEntry(ctx, snapshot.convID, snapshot.token) + continue + } if err := s.diskStorage.PutTokenEntry(ctx, snapshot.convID, snapshot.token, snapshot.entry); err != nil { s.Debug(ctx, "Flush: failed to write token: %s", err) + s.requeue(tokenSnapshots[i:], aliasSnapshots, mdSnapshots) return err } } + tokenSnapshots = nil for alias, ae := range aliasSnapshots { + if ae == nil { + s.diskStorage.RemoveAliasEntry(ctx, alias) + delete(aliasSnapshots, alias) + continue + } if err := s.diskStorage.PutAliasEntry(ctx, alias, ae); err != nil { s.Debug(ctx, "Flush: failed to write alias: %s", err) + s.requeue(nil, aliasSnapshots, mdSnapshots) return err } + delete(aliasSnapshots, alias) } for convIDStr, md := range mdSnapshots { convID, err := chat1.MakeConvID(string(convIDStr)) if err != nil { s.Debug(ctx, "Flush: invalid convID %s: %s", convIDStr, err) + delete(mdSnapshots, convIDStr) continue } if err := s.diskStorage.PutMetadata(ctx, convID, md); err != nil { s.Debug(ctx, "Flush: failed to write metadata: %s", err) + s.requeue(nil, nil, mdSnapshots) return err } + delete(mdSnapshots, convIDStr) } return nil } + +// requeue puts snapshots that never reached disk back into the pending set so a +// later flush retries them. +// +// Without this a failed write was simply dropped: the entry had already been +// removed from dirty tracking before the write was attempted, and nothing marks +// it dirty again, since nobody mutates a token entry for a message that is +// already indexed. The live metadata meanwhile keeps its SeenIDs, so the next +// successful flush would record those messages as indexed with no tokens on +// disk - unsearchable, and never re-indexed because the conv reads as complete. +// +// A key written again since the snapshot was taken is left alone: that pending +// value is newer than what failed to write. +func (s *store) requeue(tokens []tokenSnapshot, aliases map[string]*aliasEntry, + mds map[chat1.ConvIDStr]*indexMetadata, +) { + s.Lock() + defer s.Unlock() + for _, snapshot := range tokens { + convIDStr := snapshot.convID.ConvIDStr() + if s.dirtyTokens[convIDStr] == nil { + s.dirtyTokens[convIDStr] = make(map[string]*tokenEntry) + } + if _, ok := s.dirtyTokens[convIDStr][snapshot.token]; ok { + continue + } + s.dirtyTokens[convIDStr][snapshot.token] = snapshot.entry + s.dirtyCount++ + } + for alias, ae := range aliases { + if _, ok := s.dirtyAliases[alias]; ok { + continue + } + s.dirtyAliases[alias] = ae + s.dirtyCount++ + } + for convIDStr, md := range mds { + if _, ok := s.dirtyMetadata[convIDStr]; ok { + continue + } + s.dirtyMetadata[convIDStr] = md + s.dirtyCount++ + } +} From e411dadab2d43b2e6dc5665f6b8f2fdcec3e75c3 Mon Sep 17 00:00:00 2001 From: chrisnojima Date: Sun, 26 Jul 2026 15:01:47 -0400 Subject: [PATCH 2/8] perf(teams): memoize GetAnnotatedTeam The annotated team is the mostly off-chain bundle behind the team page, and it was rebuilt on every read: 469 team loads at 292ms each, 186 seconds, in a single run. It is now memoized behind an AnnotatedTeamCacher. Invalidation hangs off the notification that already exists: NotifyRouter drops the entry on any teamChangedByID, local or server-pushed, so the next UI read rebuilds from the server. The interface lives in libkb so the router can reach it; the implementation stays in teams. --- go/libkb/globals.go | 50 ++++++--- go/libkb/interfaces.go | 10 ++ go/libkb/notify_router.go | 6 ++ go/teams/annotated_cache.go | 179 +++++++++++++++++++++++++++++++ go/teams/annotated_cache_test.go | 147 +++++++++++++++++++++++++ go/teams/handler.go | 9 ++ go/teams/init.go | 1 + go/teams/service_helper.go | 16 ++- 8 files changed, 399 insertions(+), 19 deletions(-) create mode 100644 go/teams/annotated_cache.go create mode 100644 go/teams/annotated_cache_test.go diff --git a/go/libkb/globals.go b/go/libkb/globals.go index 8fb1f0814343..324fa458a9bd 100644 --- a/go/libkb/globals.go +++ b/go/libkb/globals.go @@ -91,23 +91,24 @@ type GlobalContext struct { loadUserLockTab *LockTable teamAuditor TeamAuditor teamBoxAuditor TeamBoxAuditor - stellar Stellar // Stellar related ops - deviceEKStorage DeviceEKStorage // Store device ephemeral keys - userEKBoxStorage UserEKBoxStorage // Store user ephemeral key boxes - teamEKBoxStorage TeamEKBoxStorage // Store team ephemeral key boxes - teambotEKBoxStorage TeamEKBoxStorage // Store team bot ephemeral key boxes - ekLib EKLib // Wrapper to call ephemeral key methods - teambotBotKeyer TeambotBotKeyer // TeambotKeyer for bot members - teambotMemberKeyer TeambotMemberKeyer // TeambotKeyer for non-bot members - itciCacher LRUer // Cacher for implicit team conflict info - iteamCacher MemLRUer // In memory cacher for implicit teams - cardCache *UserCardCache // cache of keybase1.UserCard objects - fullSelfer FullSelfer // a loader that gets the full self object - pvlSource MerkleStore // a cache and fetcher for pvl - paramProofStore MerkleStore // a cache and fetcher for param proofs - externalURLStore MerkleStore // a cache and fetcher for external urls - PayloadCache *PayloadCache // cache of ChainLink payload json wrappers - kvRevisionCache KVRevisionCacher // cache of revisions for verifying key-value store results + stellar Stellar // Stellar related ops + deviceEKStorage DeviceEKStorage // Store device ephemeral keys + userEKBoxStorage UserEKBoxStorage // Store user ephemeral key boxes + teamEKBoxStorage TeamEKBoxStorage // Store team ephemeral key boxes + teambotEKBoxStorage TeamEKBoxStorage // Store team bot ephemeral key boxes + ekLib EKLib // Wrapper to call ephemeral key methods + teambotBotKeyer TeambotBotKeyer // TeambotKeyer for bot members + teambotMemberKeyer TeambotMemberKeyer // TeambotKeyer for non-bot members + itciCacher LRUer // Cacher for implicit team conflict info + iteamCacher MemLRUer // In memory cacher for implicit teams + annotatedTeamCacher AnnotatedTeamCacher // In memory cacher for keybase1.AnnotatedTeam + cardCache *UserCardCache // cache of keybase1.UserCard objects + fullSelfer FullSelfer // a loader that gets the full self object + pvlSource MerkleStore // a cache and fetcher for pvl + paramProofStore MerkleStore // a cache and fetcher for param proofs + externalURLStore MerkleStore // a cache and fetcher for external urls + PayloadCache *PayloadCache // cache of ChainLink payload json wrappers + kvRevisionCache KVRevisionCacher // cache of revisions for verifying key-value store results Pegboard *Pegboard GpgClient *GpgCLI // A standard GPG-client (optional) @@ -770,6 +771,21 @@ func (g *GlobalContext) SetImplicitTeamCacher(l MemLRUer) { g.iteamCacher = l } +func (g *GlobalContext) GetAnnotatedTeamCacher() AnnotatedTeamCacher { + g.cacheMu.RLock() + defer g.cacheMu.RUnlock() + return g.annotatedTeamCacher +} + +func (g *GlobalContext) SetAnnotatedTeamCacher(c AnnotatedTeamCacher) { + // a write needs the write lock: GetAnnotatedTeamCacher runs from + // NotifyRouter.HandleTeamChangedByID on every team notification. The + // neighbouring setters take RLock here too, and are equally wrong. + g.cacheMu.Lock() + defer g.cacheMu.Unlock() + g.annotatedTeamCacher = c +} + func (g *GlobalContext) GetKVRevisionCache() KVRevisionCacher { g.cacheMu.RLock() defer g.cacheMu.RUnlock() diff --git a/go/libkb/interfaces.go b/go/libkb/interfaces.go index 9b07778fa3ab..ac941791c74f 100644 --- a/go/libkb/interfaces.go +++ b/go/libkb/interfaces.go @@ -965,6 +965,16 @@ type MemLRUer interface { OnDbNuke(mctx MetaContext) error } +// AnnotatedTeamCacher memoizes the expensive, mostly off-chain bundle of data that +// backs the team page in the UI. Implemented in the teams package; the interface lives +// here so NotifyRouter can invalidate entries when a team changes. +type AnnotatedTeamCacher interface { + Get(mctx MetaContext, teamID keybase1.TeamID) (keybase1.AnnotatedTeam, bool) + Put(mctx MetaContext, teamID keybase1.TeamID, team keybase1.AnnotatedTeam) + Remove(teamID keybase1.TeamID) + Clear() +} + type ClockContext interface { GetClock() clockwork.Clock } diff --git a/go/libkb/notify_router.go b/go/libkb/notify_router.go index 980e97f663ca..2de92e116acc 100644 --- a/go/libkb/notify_router.go +++ b/go/libkb/notify_router.go @@ -2193,6 +2193,12 @@ func (n *NotifyRouter) HandleTeamChangedByID(ctx context.Context, return } + // Any change to the team, local or server-pushed, drops the memoized + // AnnotatedTeam so the next UI read rebuilds it from the server. + if c := n.G().GetAnnotatedTeamCacher(); c != nil { + c.Remove(teamID) + } + arg := keybase1.TeamChangedByIDArg{ TeamID: teamID, LatestSeqno: latestSeqno, diff --git a/go/teams/annotated_cache.go b/go/teams/annotated_cache.go new file mode 100644 index 000000000000..a64148e15ccd --- /dev/null +++ b/go/teams/annotated_cache.go @@ -0,0 +1,179 @@ +package teams + +import ( + "context" + "sync" + "time" + + "github.com/keybase/client/go/libkb" + "github.com/keybase/client/go/protocol/keybase1" +) + +// annotatedTeamCacheTTL bounds how stale a memoized AnnotatedTeam can be when no +// invalidating event arrives. Every local mutation and every server-pushed team +// change purges the entry (see NotifyRouter.HandleTeamChangedByID and +// invalidateCaches), so this window only covers off-chain changes made elsewhere +// that we were not notified about. +const annotatedTeamCacheTTL = 10 * time.Second + +// maxAnnotatedTeamCacheWaits caps how many times a caller will wait on somebody +// else's in-flight load before doing the load itself, so a pathological stream of +// invalidations cannot livelock a request. +const maxAnnotatedTeamCacheWaits = 3 + +type annotatedTeamCacheEntry struct { + team keybase1.AnnotatedTeam + cachedAt time.Time +} + +// annotatedTeamCache memoizes GetAnnotatedTeam. GetAnnotatedTeam is expensive: a +// force-repolled team load plus four separate server round trips (team/for_user, +// team/get?showcase_only, team/access_requests, team/disable_tars), and the UI asks +// for the same team many times while a team page is open. This cache both collapses +// concurrent identical loads into one (single-flight) and serves a short-lived +// memoized copy to sequential callers. +type annotatedTeamCache struct { + sync.Mutex + entries map[keybase1.TeamID]annotatedTeamCacheEntry + // inflight[teamID] is closed when the current load for teamID finishes. + inflight map[keybase1.TeamID]chan struct{} + // gen[teamID] is bumped by Remove/Clear. A load that started before a bump does + // not get cached, since it may have read pre-change data. + gen map[keybase1.TeamID]uint64 +} + +var _ libkb.AnnotatedTeamCacher = (*annotatedTeamCache)(nil) + +func newAnnotatedTeamCache() *annotatedTeamCache { + return &annotatedTeamCache{ + entries: make(map[keybase1.TeamID]annotatedTeamCacheEntry), + inflight: make(map[keybase1.TeamID]chan struct{}), + gen: make(map[keybase1.TeamID]uint64), + } +} + +func (c *annotatedTeamCache) Get(mctx libkb.MetaContext, teamID keybase1.TeamID) (res keybase1.AnnotatedTeam, ok bool) { + c.Lock() + defer c.Unlock() + return c.getLocked(mctx, teamID) +} + +func (c *annotatedTeamCache) getLocked(mctx libkb.MetaContext, teamID keybase1.TeamID) (res keybase1.AnnotatedTeam, ok bool) { + e, ok := c.entries[teamID] + if !ok { + return res, false + } + if mctx.G().Clock().Now().Sub(e.cachedAt) >= annotatedTeamCacheTTL { + delete(c.entries, teamID) + return res, false + } + return e.team, true +} + +func (c *annotatedTeamCache) Put(mctx libkb.MetaContext, teamID keybase1.TeamID, team keybase1.AnnotatedTeam) { + c.Lock() + defer c.Unlock() + c.entries[teamID] = annotatedTeamCacheEntry{team: team, cachedAt: mctx.G().Clock().Now()} +} + +func (c *annotatedTeamCache) Remove(teamID keybase1.TeamID) { + c.Lock() + defer c.Unlock() + delete(c.entries, teamID) + c.gen[teamID]++ +} + +func (c *annotatedTeamCache) Clear() { + c.Lock() + defer c.Unlock() + for teamID := range c.entries { + c.gen[teamID]++ + } + for teamID := range c.inflight { + c.gen[teamID]++ + } + c.entries = make(map[keybase1.TeamID]annotatedTeamCacheEntry) +} + +func (c *annotatedTeamCache) OnLogout(mctx libkb.MetaContext) error { + c.Clear() + return nil +} + +func (c *annotatedTeamCache) OnDbNuke(mctx libkb.MetaContext) error { + c.Clear() + return nil +} + +type annotatedTeamLoader func(ctx context.Context, g *libkb.GlobalContext, teamID keybase1.TeamID) (keybase1.AnnotatedTeam, error) + +// load returns a memoized AnnotatedTeam if we have a fresh one, waits for an +// already-running load of the same team if there is one, and otherwise runs loader. +func (c *annotatedTeamCache) load(mctx libkb.MetaContext, teamID keybase1.TeamID, loader annotatedTeamLoader) (res keybase1.AnnotatedTeam, err error) { + for waits := 0; ; waits++ { + c.Lock() + if cached, ok := c.getLocked(mctx, teamID); ok { + c.Unlock() + mctx.Debug("annotatedTeamCache: hit for %v", teamID) + return cached, nil + } + if ch, ok := c.inflight[teamID]; ok && waits < maxAnnotatedTeamCacheWaits { + c.Unlock() + mctx.Debug("annotatedTeamCache: waiting on in-flight load for %v", teamID) + select { + case <-ch: + continue + case <-mctx.Ctx().Done(): + return res, mctx.Ctx().Err() + } + } + startedGen := c.gen[teamID] + ch := make(chan struct{}) + hasLead := false + if _, ok := c.inflight[teamID]; !ok { + c.inflight[teamID] = ch + hasLead = true + } + c.Unlock() + + // The leader must release its slot even if loader panics: the RPC + // handler recovers, but a populated inflight entry whose channel is + // never closed would make every later load of this team block until its + // ctx dies, for the life of the process. Deferred rather than inline for + // that reason; this path always returns below, so despite the enclosing + // loop it is registered at most once. + if hasLead { + defer func() { + c.Lock() + delete(c.inflight, teamID) + c.Unlock() + close(ch) + }() + } + + startedAt := mctx.G().Clock().Now() + res, err = loader(mctx.Ctx(), mctx.G(), teamID) + + c.Lock() + if err == nil && c.gen[teamID] == startedGen { + c.entries[teamID] = annotatedTeamCacheEntry{team: res, cachedAt: startedAt} + } + c.Unlock() + return res, err + } +} + +func NewAnnotatedTeamCacheAndInstall(g *libkb.GlobalContext) { + cache := newAnnotatedTeamCache() + g.SetAnnotatedTeamCacher(cache) + g.AddLogoutHook(cache, "annotatedTeamCache") + g.AddDbNukeHook(cache, "annotatedTeamCache") +} + +// ClearAnnotatedTeamCache drops the memoized AnnotatedTeam for a team, forcing the +// next read to go to the server. +func ClearAnnotatedTeamCache(g *libkb.GlobalContext, teamID keybase1.TeamID) { + if c := g.GetAnnotatedTeamCacher(); c != nil { + c.Remove(teamID) + } +} diff --git a/go/teams/annotated_cache_test.go b/go/teams/annotated_cache_test.go new file mode 100644 index 000000000000..6a3a7d0d2267 --- /dev/null +++ b/go/teams/annotated_cache_test.go @@ -0,0 +1,147 @@ +package teams + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + + "github.com/keybase/client/go/libkb" + "github.com/keybase/client/go/protocol/keybase1" + "github.com/keybase/clockwork" + "github.com/stretchr/testify/require" +) + +func annotatedCacheTestSetup(t *testing.T) (libkb.TestContext, libkb.MetaContext, *annotatedTeamCache, clockwork.FakeClock) { + tc := SetupTest(t, "annotated_cache", 1) + clock := clockwork.NewFakeClock() + tc.G.SetClock(clock) + cache, ok := tc.G.GetAnnotatedTeamCacher().(*annotatedTeamCache) + require.True(t, ok, "ServiceInit should install an annotatedTeamCache") + return tc, libkb.NewMetaContextForTest(tc), cache, clock +} + +func TestAnnotatedTeamCacheReusesWithinTTL(t *testing.T) { + tc, mctx, cache, clock := annotatedCacheTestSetup(t) + defer tc.Cleanup() + + teamID := keybase1.TeamID("aaa") + var calls int32 + loader := func(ctx context.Context, g *libkb.GlobalContext, id keybase1.TeamID) (keybase1.AnnotatedTeam, error) { + atomic.AddInt32(&calls, 1) + return keybase1.AnnotatedTeam{TeamID: id, Name: "team.one"}, nil + } + + for i := 0; i < 5; i++ { + res, err := cache.load(mctx, teamID, loader) + require.NoError(t, err) + require.Equal(t, "team.one", res.Name) + } + require.EqualValues(t, 1, atomic.LoadInt32(&calls)) + + clock.Advance(annotatedTeamCacheTTL + 1) + _, err := cache.load(mctx, teamID, loader) + require.NoError(t, err) + require.EqualValues(t, 2, atomic.LoadInt32(&calls)) +} + +func TestAnnotatedTeamCacheInvalidation(t *testing.T) { + tc, mctx, cache, _ := annotatedCacheTestSetup(t) + defer tc.Cleanup() + + teamID := keybase1.TeamID("bbb") + var calls int32 + loader := func(ctx context.Context, g *libkb.GlobalContext, id keybase1.TeamID) (keybase1.AnnotatedTeam, error) { + atomic.AddInt32(&calls, 1) + return keybase1.AnnotatedTeam{TeamID: id}, nil + } + + _, err := cache.load(mctx, teamID, loader) + require.NoError(t, err) + cache.Remove(teamID) + _, err = cache.load(mctx, teamID, loader) + require.NoError(t, err) + require.EqualValues(t, 2, atomic.LoadInt32(&calls)) + + cache.Clear() + _, err = cache.load(mctx, teamID, loader) + require.NoError(t, err) + require.EqualValues(t, 3, atomic.LoadInt32(&calls)) +} + +// A load that was in flight when the team changed must not be cached, or we would +// serve pre-change data for the whole TTL. +func TestAnnotatedTeamCacheDropsRacedLoad(t *testing.T) { + tc, mctx, cache, _ := annotatedCacheTestSetup(t) + defer tc.Cleanup() + + teamID := keybase1.TeamID("ccc") + var calls int32 + loader := func(ctx context.Context, g *libkb.GlobalContext, id keybase1.TeamID) (keybase1.AnnotatedTeam, error) { + atomic.AddInt32(&calls, 1) + cache.Remove(teamID) + return keybase1.AnnotatedTeam{TeamID: id}, nil + } + + _, err := cache.load(mctx, teamID, loader) + require.NoError(t, err) + _, ok := cache.Get(mctx, teamID) + require.False(t, ok, "a load raced by an invalidation should not be cached") + require.EqualValues(t, 1, atomic.LoadInt32(&calls)) +} + +func TestAnnotatedTeamCacheSingleFlight(t *testing.T) { + tc, mctx, cache, _ := annotatedCacheTestSetup(t) + defer tc.Cleanup() + + teamID := keybase1.TeamID("ddd") + var calls int32 + release := make(chan struct{}) + entered := make(chan struct{}, 1) + loader := func(ctx context.Context, g *libkb.GlobalContext, id keybase1.TeamID) (keybase1.AnnotatedTeam, error) { + atomic.AddInt32(&calls, 1) + entered <- struct{}{} + <-release + return keybase1.AnnotatedTeam{TeamID: id, Name: "team.four"}, nil + } + + var wg sync.WaitGroup + results := make([]keybase1.AnnotatedTeam, 8) + for i := range results { + wg.Add(1) + go func(i int) { + defer wg.Done() + res, err := cache.load(mctx, teamID, loader) + require.NoError(t, err) + results[i] = res + }(i) + } + <-entered + close(release) + wg.Wait() + + require.EqualValues(t, 1, atomic.LoadInt32(&calls)) + for _, res := range results { + require.Equal(t, "team.four", res.Name) + } +} + +func TestAnnotatedTeamCacheDoesNotCacheErrors(t *testing.T) { + tc, mctx, cache, _ := annotatedCacheTestSetup(t) + defer tc.Cleanup() + + teamID := keybase1.TeamID("eee") + var calls int32 + loadErr := errors.New("nope") + loader := func(ctx context.Context, g *libkb.GlobalContext, id keybase1.TeamID) (keybase1.AnnotatedTeam, error) { + atomic.AddInt32(&calls, 1) + return keybase1.AnnotatedTeam{}, loadErr + } + + for i := 0; i < 3; i++ { + _, err := cache.load(mctx, teamID, loader) + require.ErrorIs(t, err, loadErr) + } + require.EqualValues(t, 3, atomic.LoadInt32(&calls)) +} diff --git a/go/teams/handler.go b/go/teams/handler.go index 5ff0ab33f087..070b4a1bfd95 100644 --- a/go/teams/handler.go +++ b/go/teams/handler.go @@ -241,6 +241,7 @@ func invalidateCaches(mctx libkb.MetaContext, teamID keybase1.TeamID) { // refresh the KBFS Favorites cache since it no longer should contain // this team. mctx.G().NotifyRouter.HandleFavoritesChanged(mctx.G().GetMyUID()) + ClearAnnotatedTeamCache(mctx.G(), teamID) if ekLib := mctx.G().GetEKLib(); ekLib != nil { ekLib.PurgeTeamEKCachesForTeamID(mctx, teamID) ekLib.PurgeTeambotEKCachesForTeamID(mctx, teamID) @@ -308,6 +309,10 @@ func HandleChangeNotification(ctx context.Context, g *libkb.GlobalContext, rows func HandleTeamMemberShowcaseChange(ctx context.Context, g *libkb.GlobalContext) (err error) { defer g.CTrace(ctx, "HandleTeamMemberShowcaseChange", &err)() + // The notification does not say which team changed, so drop everything. + if c := g.GetAnnotatedTeamCacher(); c != nil { + c.Clear() + } g.NotifyRouter.HandleTeamMetadataUpdate(ctx) return nil } @@ -519,6 +524,10 @@ func HandleOpenTeamAccessRequest(ctx context.Context, g *libkb.GlobalContext, ms ctx = libkb.WithLogTag(ctx, "CLKR") defer g.CTrace(ctx, "HandleOpenTeamAccessRequest", &err)() + // Access requests are off-chain, so a membership change is not guaranteed to + // follow; drop the memoized team so the UI picks up the new request list. + ClearAnnotatedTeamCache(g, msg.TeamID) + return RetryIfPossible(ctx, g, func(ctx context.Context, _ int) error { team, err := Load(ctx, g, keybase1.LoadTeamArg{ ID: msg.TeamID, diff --git a/go/teams/init.go b/go/teams/init.go index 57f09b201e72..0d3109187827 100644 --- a/go/teams/init.go +++ b/go/teams/init.go @@ -12,6 +12,7 @@ func ServiceInit(g *libkb.GlobalContext) { NewBoxAuditorAndInstall(g) NewImplicitTeamConflictInfoCacheAndInstall(g) NewImplicitTeamCacheAndInstall(g) + NewAnnotatedTeamCacheAndInstall(g) hidden.NewChainManagerAndInstall(g) NewTeamRoleMapManagerAndInstall(g) } diff --git a/go/teams/service_helper.go b/go/teams/service_helper.go index c14647d05085..15b473648d9a 100644 --- a/go/teams/service_helper.go +++ b/go/teams/service_helper.go @@ -34,9 +34,21 @@ func LoadTeamPlusApplicationKeys(ctx context.Context, g *libkb.GlobalContext, id // GetAnnotatedTeam bundles up various data, both on and off chain, about a specific team for // consumption by the UI. In particular, it supplies almost all of the information on a team's // subpage in the Teams tab. -// It always repolls to ensure latest version of a team, but member infos (username, full name, if -// they reset or not) are subject to UIDMapper caching. +// +// The result is memoized (see annotated_cache.go): concurrent requests for the same team share +// one load, and repeat requests inside a short window reuse the previous result. Any team change +// we hear about, local or server-pushed, invalidates the entry. func GetAnnotatedTeam(ctx context.Context, g *libkb.GlobalContext, teamID keybase1.TeamID) (res keybase1.AnnotatedTeam, err error) { + cache, _ := g.GetAnnotatedTeamCacher().(*annotatedTeamCache) + if cache == nil { + return loadAnnotatedTeam(ctx, g, teamID) + } + return cache.load(libkb.NewMetaContext(ctx, g), teamID, loadAnnotatedTeam) +} + +// loadAnnotatedTeam always repolls to ensure latest version of a team, but member infos +// (username, full name, if they reset or not) are subject to UIDMapper caching. +func loadAnnotatedTeam(ctx context.Context, g *libkb.GlobalContext, teamID keybase1.TeamID) (res keybase1.AnnotatedTeam, err error) { tracer := g.CTimeTracer(ctx, "GetAnnotatedTeam", true) defer tracer.Finish() From 8bcd02345d952719746763533f7ecd90c4941ac4 Mon Sep 17 00:00:00 2001 From: chrisnojima Date: Sun, 26 Jul 2026 15:01:47 -0400 Subject: [PATCH 3/8] perf(identify): share a concurrent remote proof check instead of repeating it Several UI surfaces routinely identify the same user at the same moment - the profile screen, the tracker popup, the profile card - and each sent its own request to the third-party service. Those services rate limit, and a throttled fetch surfaces to the user as a broken proof. checkStatusShared collapses a check with an identical one already in flight. It shares work, never results: a check is eligible only if it began at or after the caller's requestedAt, so the caller gets a live check that ran during its own request, and nothing is remembered past it. This neither lengthens a cache lifetime nor lets a stale answer stand in for a new one. finish is deferred, so a panic in the checker still closes the flight - without that the entry sits in the LRU with its channel open and every session sharing that key blocks until its own context is canceled. --- go/engine/identify2_with_uid.go | 10 +++- go/libkb/id_table.go | 56 +++++++++++++++++-- go/libkb/proof_cache.go | 86 +++++++++++++++++++++++++++++ go/libkb/proof_cache_flight_test.go | 75 +++++++++++++++++++++++++ 4 files changed, 220 insertions(+), 7 deletions(-) create mode 100644 go/libkb/proof_cache_flight_test.go diff --git a/go/engine/identify2_with_uid.go b/go/engine/identify2_with_uid.go index 093c97a2b901..ec6cb5c492dc 100644 --- a/go/engine/identify2_with_uid.go +++ b/go/engine/identify2_with_uid.go @@ -252,6 +252,10 @@ type Identify2WithUID struct { resultCh chan<- error + // When this identify was asked for; bounds which in-flight remote proof + // checks it may share with concurrent identifies of the same user. + requestedAt time.Time + // For eagerly checking remote Assertions as they come in, these // member variables maintain state, protected by the remotesMutex. remotesMutex sync.Mutex @@ -330,6 +334,10 @@ func (e *Identify2WithUID) Run(m libkb.MetaContext) (err error) { return libkb.NoUIDError{} } + // Remote proof checks started from here on may be shared with other identify + // sessions, but only if they started after this point. + e.requestedAt = m.G().Clock().Now() + // Only the first send matters, but we don't want to block the subsequent no-op // sends. This code will break when we have more than 100 unblocking opportunities. ch := make(chan error, 100) @@ -884,7 +892,7 @@ func (e *Identify2WithUID) runIdentifyUI(m libkb.MetaContext) (err error) { e.metaContext = m if them.IDTable() == nil { m.Debug("| No IDTable for user") - } else if err = them.IDTable().Identify(m, e.state, e.forceRemoteCheck(), iui, e, identifyTableMode); err != nil { + } else if err = them.IDTable().Identify(m, e.state, e.forceRemoteCheck(), e.requestedAt, iui, e, identifyTableMode); err != nil { m.Debug("| Failure in running IDTable") return err } diff --git a/go/libkb/id_table.go b/go/libkb/id_table.go index 6fa557890ad5..50cb23bd60e6 100644 --- a/go/libkb/id_table.go +++ b/go/libkb/id_table.go @@ -1668,11 +1668,13 @@ const ( IdentifyTableModeActive IdentifyTableMode = iota ) -func (idt *IdentityTable) Identify(m MetaContext, is IdentifyState, forceRemoteCheck bool, ui IdentifyUI, ccl CheckCompletedListener, itm IdentifyTableMode) error { +// requestedAt is when the identify this table walk belongs to was asked for. It +// bounds which in-flight remote proof checks this walk is allowed to share. +func (idt *IdentityTable) Identify(m MetaContext, is IdentifyState, forceRemoteCheck bool, requestedAt time.Time, ui IdentifyUI, ccl CheckCompletedListener, itm IdentifyTableMode) error { errs := make(chan error, len(is.res.ProofChecks)) for _, lcr := range is.res.ProofChecks { go func(l *LinkCheckResult) { - errs <- idt.identifyActiveProof(m, l, is, forceRemoteCheck, ui, ccl, itm) + errs <- idt.identifyActiveProof(m, l, is, forceRemoteCheck, requestedAt, ui, ccl, itm) }(lcr) } @@ -1701,8 +1703,8 @@ func (idt *IdentityTable) Identify(m MetaContext, is IdentifyState, forceRemoteC // ========================================================================= -func (idt *IdentityTable) identifyActiveProof(m MetaContext, lcr *LinkCheckResult, is IdentifyState, forceRemoteCheck bool, ui IdentifyUI, ccl CheckCompletedListener, itm IdentifyTableMode) error { - idt.proofRemoteCheck(m, is.HasPreviousTrack(), forceRemoteCheck, lcr, itm) +func (idt *IdentityTable) identifyActiveProof(m MetaContext, lcr *LinkCheckResult, is IdentifyState, forceRemoteCheck bool, requestedAt time.Time, ui IdentifyUI, ccl CheckCompletedListener, itm IdentifyTableMode) error { + idt.proofRemoteCheck(m, is.HasPreviousTrack(), forceRemoteCheck, requestedAt, lcr, itm) if ccl != nil { ccl.CCLCheckCompleted(lcr) } @@ -1763,7 +1765,7 @@ func (idt *IdentityTable) ComputeRemoteDiff(tracked, trackedTmp, observed keybas return ret } -func (idt *IdentityTable) proofRemoteCheck(m MetaContext, hasPreviousTrack, forceRemoteCheck bool, res *LinkCheckResult, itm IdentifyTableMode) { +func (idt *IdentityTable) proofRemoteCheck(m MetaContext, hasPreviousTrack, forceRemoteCheck bool, requestedAt time.Time, res *LinkCheckResult, itm IdentifyTableMode) { p := res.link m.Debug("+ RemoteCheckProof %s", p.ToDebugString()) @@ -1854,7 +1856,7 @@ func (idt *IdentityTable) proofRemoteCheck(m MetaContext, hasPreviousTrack, forc if res.hint != nil { hint = *res.hint } - res.verifiedHint, res.err = pc.CheckStatus(m, hint, pcm, pvlU) + res.verifiedHint, res.err = idt.checkStatusShared(m, pc, hint, pcm, pvlU, sid, requestedAt) // If no error than all good if res.err == nil { @@ -1877,6 +1879,48 @@ func (idt *IdentityTable) proofRemoteCheck(m MetaContext, hasPreviousTrack, forc m.Debug("| Check status (%s) failed with error: %s", p.ToDebugString(), res.err.Error()) } +// checkStatusShared performs the outbound check of a remote proof, collapsing it +// with an identical check that another identify session already started after +// this one was requested. Several UI surfaces routinely identify the same user at +// the same moment (profile screen, tracker popup, profile card), and each one +// used to send its own request to the third-party service; those services rate +// limit, and a throttled fetch shows up to the user as a broken proof. +// +// This only shares work, never results: a check is eligible only if it began at +// or after requestedAt, so what the caller gets back is a live check that ran +// during its own request. Nothing is remembered past the check, so this neither +// lengthens any cache lifetime nor lets a stale answer stand in for a new one. +func (idt *IdentityTable) checkStatusShared(m MetaContext, pc ProofChecker, hint SigHint, + pcm ProofCheckerMode, pvlU keybase1.MerkleStoreEntry, sid keybase1.SigID, + requestedAt time.Time, +) (*SigHint, ProofError) { + // Everything that can change the outcome of the check has to be in the key: + // which proof, which pvl script, which checker mode (passive checks skip + // self-hosted services), and which hint we're checking against. + key := fmt.Sprintf("%s|%s|%d|%s|%s", sid, pvlU.Hash, pcm, hint.GetAPIURL(), hint.GetCheckText()) + + mine, theirs := m.G().ProofCache.CheckFlightBegin(key, requestedAt, m.G().Clock().Now()) + + if theirs != nil { + if verifiedHint, perr, usable := theirs.wait(m.Ctx()); usable { + m.Debug("| Shared an in-flight remote proof check for %s (started %s)", sid, theirs.startedAt) + return verifiedHint, perr + } + return pc.CheckStatus(m, hint, pcm, pvlU) + } + + var verifiedHint *SigHint + var perr ProofError + if mine != nil { + // Deferred so a panic in the checker still closes the flight. Otherwise + // the entry sits in the LRU with an open channel and every session that + // shares it blocks until its own context is canceled. + defer func() { mine.finish(verifiedHint, perr, m.Ctx().Err() == nil) }() + } + verifiedHint, perr = pc.CheckStatus(m, hint, pcm, pvlU) + return verifiedHint, perr +} + // VerifyReverseSig checks reverse signature using the key provided. // does not modify `payload`. // `path` is the path to the reverse sig spot to null before checking. diff --git a/go/libkb/proof_cache.go b/go/libkb/proof_cache.go index 0a75b516e281..5f0daac14e64 100644 --- a/go/libkb/proof_cache.go +++ b/go/libkb/proof_cache.go @@ -4,6 +4,7 @@ package libkb import ( + "context" "fmt" "sync" "time" @@ -111,6 +112,85 @@ type ProofCache struct { lru *lru.Cache sync.RWMutex noDisk bool + + flightMu sync.Mutex + flights *lru.Cache +} + +// proofCheckFlightCapacity bounds the singleflight table. It is a memory bound +// only: sharing is gated on when a check started, never on how long an entry has +// been kept around, so retention can't make a shared answer any staler. +const proofCheckFlightCapacity = 500 + +// proofCheckFlight is one outbound remote proof check that other identify +// sessions may share instead of issuing a duplicate request of their own. +type proofCheckFlight struct { + startedAt time.Time + doneCh chan struct{} + + // Only valid once doneCh is closed. + hint *SigHint + err ProofError + usable bool +} + +func (f *proofCheckFlight) finish(hint *SigHint, err ProofError, usable bool) { + f.hint = hint + f.err = err + f.usable = usable + close(f.doneCh) +} + +// wait blocks until the shared check finishes. usable is false when the result +// can't stand in for the caller's own check (the owning goroutine was canceled), +// in which case the caller must do its own check. +func (f *proofCheckFlight) wait(ctx context.Context) (hint *SigHint, err ProofError, usable bool) { + select { + case <-f.doneCh: + return f.hint, f.err, f.usable + case <-ctx.Done(): + return nil, nil, false + } +} + +// CheckFlightBegin either registers the caller as the one that will perform the +// outbound proof check for key (returning mine), or hands back a check another +// session already has going that can answer for the caller (returning theirs). +// +// requestedAt is when the caller asked for this check. A check may only be +// shared if it *started* at or after that moment, so the shared answer is never +// older than one the caller would have produced by doing the work itself. That +// makes this a pure singleflight: it collapses duplicate concurrent work without +// letting any result outlive the freshness its requester asked for. +func (pc *ProofCache) CheckFlightBegin(key string, requestedAt, now time.Time) (mine, theirs *proofCheckFlight) { + if pc == nil { + return nil, nil + } + + pc.flightMu.Lock() + defer pc.flightMu.Unlock() + + if pc.flights == nil { + l, err := lru.New(proofCheckFlightCapacity) + if err != nil { + return nil, nil + } + pc.flights = l + } + + // A zero requestedAt would make every entry look eligible, so callers that + // can't say when they asked never share. + if !requestedAt.IsZero() { + if tmp, found := pc.flights.Get(key); found { + if f, ok := tmp.(*proofCheckFlight); ok && !f.startedAt.Before(requestedAt) { + return nil, f + } + } + } + + f := &proofCheckFlight{startedAt: now, doneCh: make(chan struct{})} + pc.flights.Add(key, f) + return f, nil } func NewProofCache(g *GlobalContext, capac int) *ProofCache { @@ -124,6 +204,12 @@ func (pc *ProofCache) DisableDisk() { } func (pc *ProofCache) Reset() error { + pc.flightMu.Lock() + if pc.flights != nil { + pc.flights.Purge() + } + pc.flightMu.Unlock() + pc.Lock() defer pc.Unlock() return pc.initCache() diff --git a/go/libkb/proof_cache_flight_test.go b/go/libkb/proof_cache_flight_test.go new file mode 100644 index 000000000000..85bbcb9bbccf --- /dev/null +++ b/go/libkb/proof_cache_flight_test.go @@ -0,0 +1,75 @@ +package libkb + +import ( + "context" + "testing" + "time" +) + +func TestProofCheckFlightSharesConcurrentCheck(t *testing.T) { + pc := NewProofCache(nil, 10) + t0 := time.Now() + + // Two sessions ask at the same moment; the first one starts the check. + mine, theirs := pc.CheckFlightBegin("k", t0, t0.Add(time.Millisecond)) + if mine == nil || theirs != nil { + t.Fatalf("first caller should own the flight, got mine=%v theirs=%v", mine, theirs) + } + + mine2, theirs2 := pc.CheckFlightBegin("k", t0, t0.Add(2*time.Millisecond)) + if mine2 != nil || theirs2 != mine { + t.Fatalf("second caller should share the first flight") + } + + hint := NewVerifiedSigHint("", "remote", "api", "human", "text") + go mine.finish(hint, nil, true) + + gotHint, gotErr, usable := theirs2.wait(context.Background()) + if !usable || gotErr != nil || gotHint != hint { + t.Fatalf("shared result mismatch: %v %v %v", gotHint, gotErr, usable) + } +} + +func TestProofCheckFlightNotSharedWithLaterRequest(t *testing.T) { + pc := NewProofCache(nil, 10) + t0 := time.Now() + + mine, _ := pc.CheckFlightBegin("k", t0, t0) + mine.finish(nil, nil, true) + + // A request made after that check started must not be answered by it. + mine2, theirs2 := pc.CheckFlightBegin("k", t0.Add(time.Nanosecond), t0.Add(time.Second)) + if theirs2 != nil || mine2 == nil { + t.Fatalf("a later request must run its own check") + } +} + +func TestProofCheckFlightZeroRequestedAtNeverShares(t *testing.T) { + pc := NewProofCache(nil, 10) + t0 := time.Now() + + mine, _ := pc.CheckFlightBegin("k", t0, t0) + mine.finish(nil, nil, true) + + mine2, theirs2 := pc.CheckFlightBegin("k", time.Time{}, t0) + if theirs2 != nil || mine2 == nil { + t.Fatalf("a caller with no request time must run its own check") + } +} + +func TestProofCheckFlightUnusableResultFallsThrough(t *testing.T) { + pc := NewProofCache(nil, 10) + t0 := time.Now() + + mine, _ := pc.CheckFlightBegin("k", t0, t0) + // Owner was canceled, so its result can't stand in for anyone else. + mine.finish(nil, nil, false) + + _, theirs := pc.CheckFlightBegin("k", t0, t0) + if theirs == nil { + t.Fatal("expected to find the flight") + } + if _, _, usable := theirs.wait(context.Background()); usable { + t.Fatal("canceled result must not be usable") + } +} From e1fce9315fb95f664207c467c6b3121d269a0100 Mon Sep 17 00:00:00 2001 From: chrisnojima Date: Sun, 26 Jul 2026 15:01:47 -0400 Subject: [PATCH 4/8] perf(chat): cut the per-item cost out of three service-side loops Three places did per-item work inside a loop over a whole set. userEmojis read the gregor state once per emoji to decide whether animations were disabled; the state is now read once for the set and passed in, and the second attachment-URL lookup is skipped when it would return the same URL as the first. The search indexer fetched superseded messages one at a time - 14,621 single-message GetMessages calls in one run - where a batch of edits and attachment uploads can be fetched in one call. The per-message path stays as the fallback, since a failed batch says nothing about which id was at fault. Conversation participants are served from a five minute cache rather than refreshed per request, and uidmap returns the full name package it found on disk instead of dropping the part the caller asked for. --- go/chat/emojisource.go | 23 +++++++++++++++++++--- go/chat/participantsource.go | 37 ++++++++++++++++++++++++++++++++---- go/uidmap/uidmap.go | 1 + 3 files changed, 54 insertions(+), 7 deletions(-) diff --git a/go/chat/emojisource.go b/go/chat/emojisource.go index c88b27e77521..1867d8b6f344 100644 --- a/go/chat/emojisource.go +++ b/go/chat/emojisource.go @@ -494,16 +494,28 @@ func (s *DevConvEmojiSource) ToggleAnimations(ctx context.Context, uid gregor1.U func (s *DevConvEmojiSource) RemoteToLocalSource(ctx context.Context, uid gregor1.UID, remote chat1.EmojiRemoteSource, +) (source chat1.EmojiLoadSource, noAnimSource chat1.EmojiLoadSource, err error) { + return s.remoteToLocalSource(ctx, remote, s.animationsDisabled(ctx, uid)) +} + +// callers converting a whole set of emojis pass noAnim in so the gregor state is +// read once for the set instead of once per emoji +func (s *DevConvEmojiSource) remoteToLocalSource(ctx context.Context, remote chat1.EmojiRemoteSource, + noAnim bool, ) (source chat1.EmojiLoadSource, noAnimSource chat1.EmojiLoadSource, err error) { typ, err := remote.Typ() if err != nil { return source, noAnimSource, err } - noAnim := s.animationsDisabled(ctx, uid) switch typ { case chat1.EmojiRemoteSourceTyp_MESSAGE: msg := remote.Message() sourceURL := s.G().AttachmentURLSrv.GetURL(ctx, msg.ConvID, msg.MsgID, false, noAnim, true) + if noAnim { + // same arguments, so the second lookup would return the same URL + ret := chat1.NewEmojiLoadSourceWithHttpsrv(sourceURL) + return ret, ret, nil + } noAnimSourceURL := s.G().AttachmentURLSrv.GetURL(ctx, msg.ConvID, msg.MsgID, false, true, true) return chat1.NewEmojiLoadSourceWithHttpsrv(sourceURL), chat1.NewEmojiLoadSourceWithHttpsrv(noAnimSourceURL), nil @@ -573,6 +585,9 @@ func (s *DevConvEmojiSource) getNoSet(ctx context.Context, uid gregor1.UID, conv } convs := ibox.Convs seenAliases := make(map[string]int) + // read once for the whole set: this runs for every emoji the user has, and each + // read is a full gregor state fetch + noAnim := s.animationsDisabled(ctx, uid) addEmojis := func(convs []chat1.ConversationLocal, isCrossTeam bool) { if opts.OnlyInTeam && isCrossTeam { return @@ -596,7 +611,7 @@ func (s *DevConvEmojiSource) getNoSet(ctx context.Context, uid gregor1.UID, conv continue } var creationInfo *chat1.EmojiCreationInfo - source, noAnimSource, err := s.RemoteToLocalSource(ctx, uid, storedEmoji) + source, noAnimSource, err := s.remoteToLocalSource(ctx, storedEmoji, noAnim) if err != nil { s.Debug(ctx, "Get: skipping emoji on remote-to-local error: %s", err) continue @@ -954,9 +969,11 @@ func (s *DevConvEmojiSource) Decorate(ctx context.Context, body string, uid greg offset := 0 added := 0 isReacji := messageType == chat1.MessageType_REACTION + // read once for the whole message rather than once per matched emoji + noAnim := s.animationsDisabled(ctx, uid) for _, match := range matches { if remoteSource, ok := emojiMap[match.name]; ok { - source, noAnimSource, err := s.RemoteToLocalSource(ctx, uid, remoteSource) + source, noAnimSource, err := s.remoteToLocalSource(ctx, remoteSource, noAnim) if err != nil { s.Debug(ctx, "Decorate: failed to get local source: %s", err) continue diff --git a/go/chat/participantsource.go b/go/chat/participantsource.go index 07156ce630ce..0bf2df75b3b9 100644 --- a/go/chat/participantsource.go +++ b/go/chat/participantsource.go @@ -18,10 +18,24 @@ import ( ) type partDiskStorage struct { - Uids []gregor1.UID - Hash string + Uids []gregor1.UID + Hash string + CachedAt time.Time } +// How long a cached participant list is served without asking the server. Every +// localization of a team conversation asks for participants, so a team screen +// with N channels used to mean N remote round trips every time it loaded, even +// when nothing had changed and the server only answered HashMatch. Membership +// changes arrive by server push for anything that re-localizes the conversation. +// +// Known gap: nothing deletes the cached entry, and every production caller +// passes DataSourceAll (the RemoteOnly path exists but is unused), so a +// membership change that does not re-localize can take up to this long to show +// up in a participant list or @-mention completion. Before the cache that was +// one cheap HashMatch round trip per localization instead. +const participantsCacheFreshness = 5 * time.Minute + type CachingParticipantSource struct { globals.Contextified utils.DebugLabeler @@ -130,6 +144,10 @@ func (s *CachingParticipantSource) GetNonblock(ctx context.Context, uid gregor1. if found { resCh <- types.ParticipantResult{Uids: local.Uids} localHash = local.Hash + if !local.CachedAt.IsZero() && + s.G().GetClock().Since(local.CachedAt) < participantsCacheFreshness { + return + } } default: } @@ -147,12 +165,23 @@ func (s *CachingParticipantSource) GetNonblock(ctx context.Context, uid gregor1. } if partRes.HashMatch { s.Debug(ctx, "GetNonblock: hash match on remote, all done") + // nothing changed, but record that we just checked so the next + // localization of this conv can be served from disk + var local partDiskStorage + found, err := s.encryptedDB.Get(ctx, s.dbKey(uid, conv.GetConvID()), &local) + if err == nil && found { + local.CachedAt = s.G().GetClock().Now() + if err := s.encryptedDB.Put(ctx, s.dbKey(uid, conv.GetConvID()), local); err != nil { + s.Debug(ctx, "GetNonblock: failed to record refresh time: %s", err) + } + } return } if err := s.encryptedDB.Put(ctx, s.dbKey(uid, conv.GetConvID()), partDiskStorage{ - Uids: partRes.Uids, - Hash: partRes.Hash, + Uids: partRes.Uids, + Hash: partRes.Hash, + CachedAt: s.G().GetClock().Now(), }); err != nil { s.Debug(ctx, "GetNonblock: failed to store participants: %s", err) } diff --git a/go/uidmap/uidmap.go b/go/uidmap/uidmap.go index eedca6258c56..d2753630af70 100644 --- a/go/uidmap/uidmap.go +++ b/go/uidmap/uidmap.go @@ -139,6 +139,7 @@ func (u *UIDMap) findFullNameLocally(ctx context.Context, g libkb.UIDMapperConte } u.fullNameCache.Add(uid, tmp) + ret = &tmp return ret, foundOnDisk } From cd6db7d9bed479543e561491a2259a132058f905 Mon Sep 17 00:00:00 2001 From: chrisnojima Date: Sun, 26 Jul 2026 15:05:08 -0400 Subject: [PATCH 5/8] test(search): make the shutdown-drain test deterministic The test queued an op, closed stopCh and called storageLoop. Both of that select's cases are ready at that point and Go chooses between them at random, so half the time the loop ran the op instead of draining it and the callback correctly reported success - which the assertion read as a failure. It passed only because the earlier version ignored the error. The drain is now its own method and the test calls it directly, so the path under test is the one that runs. 8 consecutive runs and 3 under -race are clean. --- go/chat/search/indexer.go | 39 ++++++++++++++++++++++-------------- go/chat/search/stall_test.go | 7 ++++--- 2 files changed, 28 insertions(+), 18 deletions(-) diff --git a/go/chat/search/indexer.go b/go/chat/search/indexer.go index c84786af6626..9887df649bd1 100644 --- a/go/chat/search/indexer.go +++ b/go/chat/search/indexer.go @@ -423,6 +423,28 @@ func releaseStorageCB(cb chan error, err error) { close(cb) } +// drainStorageQueue releases every op still queued at shutdown. They will not +// run, so each callback reports errStorageStopped: a caller told an op finished +// marks its messages as indexed, and nothing put them there. +// +// Split out of storageLoop because that select has both cases ready once stopCh +// closes and Go picks between them at random, which is untestable from outside. +func (idx *Indexer) drainStorageQueue() { + for { + select { + case iop := <-idx.storageCh: + switch op := iop.(type) { + case storageAdd: + releaseStorageCB(op.cb, errStorageStopped) + case storageRemove: + releaseStorageCB(op.cb, errStorageStopped) + } + default: + return + } + } +} + func (idx *Indexer) storageLoop(stopCh chan struct{}) error { ctx := context.Background() idx.Debug(ctx, "storageLoop: starting") @@ -430,21 +452,8 @@ func (idx *Indexer) storageLoop(stopCh chan struct{}) error { select { case <-stopCh: idx.Debug(ctx, "storageLoop: shutting down") - // Release anything still queued. These ops will not run, but a - // caller blocked on one's callback would never wake up otherwise. - for { - select { - case iop := <-idx.storageCh: - switch op := iop.(type) { - case storageAdd: - releaseStorageCB(op.cb, errStorageStopped) - case storageRemove: - releaseStorageCB(op.cb, errStorageStopped) - } - default: - return nil - } - } + idx.drainStorageQueue() + return nil case iop := <-idx.storageCh: switch op := iop.(type) { case storageAdd: diff --git a/go/chat/search/stall_test.go b/go/chat/search/stall_test.go index 90a487750a4c..37d7ef796501 100644 --- a/go/chat/search/stall_test.go +++ b/go/chat/search/stall_test.go @@ -224,12 +224,13 @@ func TestDroppedStorageOpDoesNotStrandCaller(t *testing.T) { // and a conv that reaches numMissing 0 that way is never revisited. func TestStorageLoopReleasesQueuedCallbacksOnShutdown(t *testing.T) { idx, _ := setupStallTestIndexer(t, "dispatch-shutdown") - stopCh := make(chan struct{}) cb := make(chan error, 1) idx.storageCh <- storageAdd{cb: cb} - close(stopCh) - require.NoError(t, idx.storageLoop(stopCh)) + // the drain directly, not storageLoop: with stopCh closed both of its select + // cases are ready and Go picks at random, so going through the loop would + // run the op half the time and test nothing the other half + idx.drainStorageQueue() select { case err := <-cb: From 0fdd17902119ee501c44c4b58a148ca10da795f2 Mon Sep 17 00:00:00 2001 From: chrisnojima Date: Sun, 26 Jul 2026 19:20:32 -0400 Subject: [PATCH 6/8] fix(libkb): take the write lock in the remaining cache setters SetImplicitTeamConflictInfoCacher, SetImplicitTeamCacher and SetKVRevisionCache mutated GlobalContext fields while holding only cacheMu.RLock(), which races with concurrent getters. Also document why the annotated team cache ages entries from the load start rather than the load finish. --- go/libkb/globals.go | 15 ++++++--------- go/teams/annotated_cache.go | 3 +++ 2 files changed, 9 insertions(+), 9 deletions(-) diff --git a/go/libkb/globals.go b/go/libkb/globals.go index 324fa458a9bd..f65272313af5 100644 --- a/go/libkb/globals.go +++ b/go/libkb/globals.go @@ -754,8 +754,8 @@ func (g *GlobalContext) GetImplicitTeamConflictInfoCacher() LRUer { } func (g *GlobalContext) SetImplicitTeamConflictInfoCacher(l LRUer) { - g.cacheMu.RLock() - defer g.cacheMu.RUnlock() + g.cacheMu.Lock() + defer g.cacheMu.Unlock() g.itciCacher = l } @@ -766,8 +766,8 @@ func (g *GlobalContext) GetImplicitTeamCacher() MemLRUer { } func (g *GlobalContext) SetImplicitTeamCacher(l MemLRUer) { - g.cacheMu.RLock() - defer g.cacheMu.RUnlock() + g.cacheMu.Lock() + defer g.cacheMu.Unlock() g.iteamCacher = l } @@ -778,9 +778,6 @@ func (g *GlobalContext) GetAnnotatedTeamCacher() AnnotatedTeamCacher { } func (g *GlobalContext) SetAnnotatedTeamCacher(c AnnotatedTeamCacher) { - // a write needs the write lock: GetAnnotatedTeamCacher runs from - // NotifyRouter.HandleTeamChangedByID on every team notification. The - // neighbouring setters take RLock here too, and are equally wrong. g.cacheMu.Lock() defer g.cacheMu.Unlock() g.annotatedTeamCacher = c @@ -793,8 +790,8 @@ func (g *GlobalContext) GetKVRevisionCache() KVRevisionCacher { } func (g *GlobalContext) SetKVRevisionCache(kvr KVRevisionCacher) { - g.cacheMu.RLock() - defer g.cacheMu.RUnlock() + g.cacheMu.Lock() + defer g.cacheMu.Unlock() g.kvRevisionCache = kvr } diff --git a/go/teams/annotated_cache.go b/go/teams/annotated_cache.go index a64148e15ccd..5d4c63ee8f1f 100644 --- a/go/teams/annotated_cache.go +++ b/go/teams/annotated_cache.go @@ -151,6 +151,9 @@ func (c *annotatedTeamCache) load(mctx libkb.MetaContext, teamID keybase1.TeamID }() } + // Age the entry from when the load started, not when it finished: the + // data describes the server state as of the request, so a slow loader + // must not buy the result extra TTL it has already spent being stale. startedAt := mctx.G().Clock().Now() res, err = loader(mctx.Ctx(), mctx.G(), teamID) From 64bac6dadb9d9bbd508383849dc00fe53c1197dc Mon Sep 17 00:00:00 2001 From: chrisnojima Date: Mon, 27 Jul 2026 16:51:43 -0400 Subject: [PATCH 7/8] fix(libkb): drop finished unusable proof-check flights A flight whose owner was canceled finished with usable=false but stayed in the LRU. Every later caller then got that dead flight back, fell through to its own CheckStatus, and never registered a new flight - so concurrent checks stopped collapsing, which is the whole point of the table. Remove such an entry on lookup so the next caller leads a flight others can share. Also export ProofCheckFlight (revive unexported-return) and drop the int->MessageID conversions in the flush tests (gosec G115). --- go/chat/search/flush_test.go | 8 +++---- go/libkb/proof_cache.go | 35 ++++++++++++++++++++++------- go/libkb/proof_cache_flight_test.go | 26 ++++++++++++++++----- 3 files changed, 51 insertions(+), 18 deletions(-) diff --git a/go/chat/search/flush_test.go b/go/chat/search/flush_test.go index d14295367ffa..70dcece61f59 100644 --- a/go/chat/search/flush_test.go +++ b/go/chat/search/flush_test.go @@ -246,9 +246,9 @@ func TestFlushSignalledWhenPendingSetIsFull(t *testing.T) { ctx, s, _, convID := setupFlushTestStore(t, "flush-size-bound") s.Lock() - for i := 0; i < maxDirtyEntries-1; i++ { + for i := chat1.MessageID(0); i < maxDirtyEntries-1; i++ { te := newTokenEntry() - te.MsgIDs[chat1.MessageID(i)] = chat1.EmptyStruct{} + te.MsgIDs[i] = chat1.EmptyStruct{} require.NoError(t, s.putTokenEntry(ctx, convID, fmt.Sprintf("tok%d", i), te)) } s.Unlock() @@ -276,9 +276,9 @@ func TestPendingCountTracksDistinctEntries(t *testing.T) { ctx, s, _, convID := setupFlushTestStore(t, "flush-size-distinct") s.Lock() - for i := 0; i < maxDirtyEntries*2; i++ { + for i := chat1.MessageID(0); i < maxDirtyEntries*2; i++ { te := newTokenEntry() - te.MsgIDs[chat1.MessageID(i)] = chat1.EmptyStruct{} + te.MsgIDs[i] = chat1.EmptyStruct{} require.NoError(t, s.putTokenEntry(ctx, convID, "same", te)) } count := s.dirtyCount diff --git a/go/libkb/proof_cache.go b/go/libkb/proof_cache.go index 5f0daac14e64..702492bbf6e3 100644 --- a/go/libkb/proof_cache.go +++ b/go/libkb/proof_cache.go @@ -122,9 +122,9 @@ type ProofCache struct { // been kept around, so retention can't make a shared answer any staler. const proofCheckFlightCapacity = 500 -// proofCheckFlight is one outbound remote proof check that other identify +// ProofCheckFlight is one outbound remote proof check that other identify // sessions may share instead of issuing a duplicate request of their own. -type proofCheckFlight struct { +type ProofCheckFlight struct { startedAt time.Time doneCh chan struct{} @@ -134,17 +134,28 @@ type proofCheckFlight struct { usable bool } -func (f *proofCheckFlight) finish(hint *SigHint, err ProofError, usable bool) { +func (f *ProofCheckFlight) finish(hint *SigHint, err ProofError, usable bool) { f.hint = hint f.err = err f.usable = usable close(f.doneCh) } +// finished reports whether the check is done. Only meaningful under flightMu, +// where it gates reuse of the flight's result fields. +func (f *ProofCheckFlight) finished() bool { + select { + case <-f.doneCh: + return true + default: + return false + } +} + // wait blocks until the shared check finishes. usable is false when the result // can't stand in for the caller's own check (the owning goroutine was canceled), // in which case the caller must do its own check. -func (f *proofCheckFlight) wait(ctx context.Context) (hint *SigHint, err ProofError, usable bool) { +func (f *ProofCheckFlight) wait(ctx context.Context) (hint *SigHint, err ProofError, usable bool) { select { case <-f.doneCh: return f.hint, f.err, f.usable @@ -162,7 +173,7 @@ func (f *proofCheckFlight) wait(ctx context.Context) (hint *SigHint, err ProofEr // older than one the caller would have produced by doing the work itself. That // makes this a pure singleflight: it collapses duplicate concurrent work without // letting any result outlive the freshness its requester asked for. -func (pc *ProofCache) CheckFlightBegin(key string, requestedAt, now time.Time) (mine, theirs *proofCheckFlight) { +func (pc *ProofCache) CheckFlightBegin(key string, requestedAt, now time.Time) (mine, theirs *ProofCheckFlight) { if pc == nil { return nil, nil } @@ -182,13 +193,21 @@ func (pc *ProofCache) CheckFlightBegin(key string, requestedAt, now time.Time) ( // can't say when they asked never share. if !requestedAt.IsZero() { if tmp, found := pc.flights.Get(key); found { - if f, ok := tmp.(*proofCheckFlight); ok && !f.startedAt.Before(requestedAt) { - return nil, f + if f, ok := tmp.(*ProofCheckFlight); ok && !f.startedAt.Before(requestedAt) { + // A finished flight whose owner was canceled can't answer for + // anyone, so keeping it would make every later caller fall back + // to its own check without ever forming a new flight. Drop it + // and let this caller lead one that others can share. + if f.finished() && !f.usable { + pc.flights.Remove(key) + } else { + return nil, f + } } } } - f := &proofCheckFlight{startedAt: now, doneCh: make(chan struct{})} + f := &ProofCheckFlight{startedAt: now, doneCh: make(chan struct{})} pc.flights.Add(key, f) return f, nil } diff --git a/go/libkb/proof_cache_flight_test.go b/go/libkb/proof_cache_flight_test.go index 85bbcb9bbccf..84d2405f409a 100644 --- a/go/libkb/proof_cache_flight_test.go +++ b/go/libkb/proof_cache_flight_test.go @@ -57,7 +57,7 @@ func TestProofCheckFlightZeroRequestedAtNeverShares(t *testing.T) { } } -func TestProofCheckFlightUnusableResultFallsThrough(t *testing.T) { +func TestProofCheckFlightUnusableResultDropped(t *testing.T) { pc := NewProofCache(nil, 10) t0 := time.Now() @@ -65,11 +65,25 @@ func TestProofCheckFlightUnusableResultFallsThrough(t *testing.T) { // Owner was canceled, so its result can't stand in for anyone else. mine.finish(nil, nil, false) - _, theirs := pc.CheckFlightBegin("k", t0, t0) - if theirs == nil { - t.Fatal("expected to find the flight") + // The dead flight is dropped so the next caller leads a fresh one... + mine2, theirs2 := pc.CheckFlightBegin("k", t0, t0) + if theirs2 != nil { + t.Fatal("must not share a finished unusable flight") } - if _, _, usable := theirs.wait(context.Background()); usable { - t.Fatal("canceled result must not be usable") + if mine2 == nil { + t.Fatal("expected to own a new flight") + } + + // ...which concurrent callers can then share. + _, theirs3 := pc.CheckFlightBegin("k", t0, t0) + if theirs3 != mine2 { + t.Fatal("expected to share the new flight") + } + + hint := &SigHint{apiURL: "https://example.com/proof"} + go mine2.finish(hint, nil, true) + got, _, usable := theirs3.wait(context.Background()) + if !usable || got != hint { + t.Fatalf("expected shared usable result, got usable=%v hint=%v", usable, got) } } From 5fc75eef28699ac4bbbf9205ba2077abd90f7428 Mon Sep 17 00:00:00 2001 From: chrisnojima Date: Mon, 27 Jul 2026 16:57:47 -0400 Subject: [PATCH 8/8] fix(chat): address review feedback on the caching pass - Invalidate the participant cache from the membership push instead of documenting the staleness window as a known gap. A join or leave now shows up at the next read rather than after the freshness window. - storageDispatch refuses to queue once the indexer is stopping; ops accepted after the drain had nobody left to release their callbacks. - Give storage ops a common interface so the shutdown drain does not repeat a case per op type. - Break annotatedTeamCache.load into beginLoad/endLoad/storeLoad so each locked section is a plain lock/defer unlock. - Regression tests for MarkSeen: convergence to zero missing IDs, and survival of a metadata eviction. - Comment fixes: drop an unfounded claim about team conv sharing, take the reviewer's clearer wording for the delete-vs-flush race, explain why requeue keeps a newer pending value over a failed snapshot. --- go/chat/participantsource.go | 22 +++++++--- go/chat/push.go | 38 +++++++++++++++++ go/chat/search/flush_test.go | 52 +++++++++++++++++++++++ go/chat/search/indexer.go | 43 +++++++++++++------ go/chat/search/storage.go | 17 +++++--- go/chat/types/interfaces.go | 3 ++ go/chat/types/types.go | 4 ++ go/teams/annotated_cache.go | 82 ++++++++++++++++++++++++------------ 8 files changed, 211 insertions(+), 50 deletions(-) diff --git a/go/chat/participantsource.go b/go/chat/participantsource.go index 0bf2df75b3b9..6b7bd9259134 100644 --- a/go/chat/participantsource.go +++ b/go/chat/participantsource.go @@ -29,11 +29,8 @@ type partDiskStorage struct { // when nothing had changed and the server only answered HashMatch. Membership // changes arrive by server push for anything that re-localizes the conversation. // -// Known gap: nothing deletes the cached entry, and every production caller -// passes DataSourceAll (the RemoteOnly path exists but is unused), so a -// membership change that does not re-localize can take up to this long to show -// up in a participant list or @-mention completion. Before the cache that was -// one cheap HashMatch round trip per localization instead. +// A membership push drops the entry for the convs it touches (see Invalidate), +// so this bounds only changes that arrive by neither push nor re-localization. const participantsCacheFreshness = 5 * time.Minute type CachingParticipantSource struct { @@ -104,6 +101,21 @@ func (s *CachingParticipantSource) dbKey(uid gregor1.UID, convID chat1.Conversat } } +// Invalidate drops the cached list for these convs. Membership pushes call it, +// so a join or leave shows up in participant lists and @-mention completion at +// the next read instead of waiting out participantsCacheFreshness. +func (s *CachingParticipantSource) Invalidate(ctx context.Context, uid gregor1.UID, + convIDs []chat1.ConversationID, +) { + for _, convID := range convIDs { + lock := s.locktab.AcquireOnName(ctx, s.G(), convID.String()) + if err := s.encryptedDB.Delete(ctx, s.dbKey(uid, convID)); err != nil { + s.Debug(ctx, "Invalidate: failed to delete participants for %s: %s", convID, err) + } + lock.Release(ctx) + } +} + func (s *CachingParticipantSource) GetNonblock(ctx context.Context, uid gregor1.UID, convID chat1.ConversationID, dataSource types.InboxSourceDataSourceTyp, ) (resCh chan types.ParticipantResult) { diff --git a/go/chat/push.go b/go/chat/push.go index 4107f9818082..ffd8dd8a9b30 100644 --- a/go/chat/push.go +++ b/go/chat/push.go @@ -864,6 +864,43 @@ func (g *PushHandler) notifyReset(ctx context.Context, uid gregor1.UID, g.G().ActivityNotifier.ResetConversation(ctx, uid, convID, topicType) } +// invalidateParticipants drops the cached participant list for every conv whose +// membership just changed. Without this the cache serves the pre-change list +// until it expires, which for a conv that never re-localizes is the whole +// freshness window. +func (g *PushHandler) invalidateParticipants(ctx context.Context, uid gregor1.UID, + membersRes types.MembershipUpdateRes, +) { + seen := make(map[chat1.ConvIDStr]struct{}) + var convIDs []chat1.ConversationID + add := func(convID chat1.ConversationID) { + convIDStr := convID.ConvIDStr() + if _, ok := seen[convIDStr]; ok { + return + } + seen[convIDStr] = struct{}{} + convIDs = append(convIDs, convID) + } + for _, conv := range membersRes.RoleUpdates { + add(conv.GetConvID()) + } + for _, conv := range membersRes.UserJoinedConvs { + add(conv.GetConvID()) + } + for _, cms := range [][]chat1.ConversationMember{ + membersRes.UserRemovedConvs, membersRes.UserResetConvs, + membersRes.OthersJoinedConvs, membersRes.OthersRemovedConvs, membersRes.OthersResetConvs, + } { + for _, cm := range cms { + add(cm.ConvID) + } + } + if len(convIDs) == 0 { + return + } + g.G().ParticipantsSource.Invalidate(ctx, uid, convIDs) +} + func (g *PushHandler) notifyMembersUpdate(ctx context.Context, uid gregor1.UID, membersRes types.MembershipUpdateRes, ) { @@ -1053,6 +1090,7 @@ func (g *PushHandler) MembershipUpdate(ctx context.Context, m gregor.OutOfBandMe for _, c := range updateRes.UserResetConvs { g.notifyReset(ctx, uid, c.ConvID, c.TopicType) } + g.invalidateParticipants(ctx, uid, updateRes) g.notifyMembersUpdate(ctx, uid, updateRes) g.notifyConvUpdates(ctx, uid, updateRes.RoleUpdates) diff --git a/go/chat/search/flush_test.go b/go/chat/search/flush_test.go index 70dcece61f59..0a7d04af1bf4 100644 --- a/go/chat/search/flush_test.go +++ b/go/chat/search/flush_test.go @@ -458,6 +458,58 @@ func TestAddDoesNotMarkSeenWhenIndexingFails(t *testing.T) { "message was marked indexed even though indexing it failed") } +// IDs the server will never return - deleted messages, gaps that never existed - +// cannot be closed by indexing them, so before MarkSeen they held numMissing +// above zero forever and SelectiveSync re-fetched the same conv every interval. +func TestMarkSeenClosesUnfetchableIDs(t *testing.T) { + ctx, s, _, convID := setupFlushTestStore(t, "mark-seen-converges") + + conv := chat1.Conversation{ + Metadata: chat1.ConversationMetadata{ConversationID: convID}, + MaxMsgSummaries: []chat1.MessageSummary{ + {MsgID: 3, MessageType: chat1.MessageType_TEXT}, + }, + } + + require.NoError(t, s.Add(ctx, convID, []chat1.MessageUnboxed{ + textMsgForTest(1, "hello world"), + textMsgForTest(3, "hello world"), + })) + + missing, err := s.MissingIDForConv(ctx, conv) + require.NoError(t, err) + require.Equal(t, []chat1.MessageID{2}, missing, + "the ID the fetch could not produce must read as missing until marked") + + // The fetch that asked for 2 succeeded and did not return it, so nothing + // else is coming for that ID. + require.NoError(t, s.MarkSeen(ctx, convID, []chat1.MessageID{2})) + + missing, err = s.MissingIDForConv(ctx, conv) + require.NoError(t, err) + require.Empty(t, missing, "a marked ID must not keep the conv incomplete") + + fullyIndexed, err := s.FullyIndexed(ctx, conv) + require.NoError(t, err) + require.True(t, fullyIndexed, "conv must converge once every ID is accounted for") +} + +// MarkSeen goes through the pending set like every other metadata mutation; a +// write-through would be undone by a concurrent flush's snapshot. +func TestMarkSeenSurvivesEviction(t *testing.T) { + ctx, s, _, convID := setupFlushTestStore(t, "mark-seen-eviction") + + require.NoError(t, s.MarkSeen(ctx, convID, []chat1.MessageID{9})) + s.mdCache.Purge() + + s.RLock() + md, err := s.getMetadataLocked(ctx, convID) + s.RUnlock() + require.NoError(t, err) + require.Contains(t, md.SeenIDs, chat1.MessageID(9), + "the mark was lost once the metadata left the cache") +} + func textMsgForTest(id chat1.MessageID, body string) chat1.MessageUnboxed { return chat1.NewMessageUnboxedWithValid(chat1.MessageUnboxedValid{ ClientHeader: chat1.MessageClientHeaderVerified{ diff --git a/go/chat/search/indexer.go b/go/chat/search/indexer.go index 9887df649bd1..338d6c08b926 100644 --- a/go/chat/search/indexer.go +++ b/go/chat/search/indexer.go @@ -23,6 +23,12 @@ import ( // is read, a prerequisite for searching. const minPriorityScore = 10 +// storageOp is what every queued storage operation has in common: a waiter that +// must be released whether or not the op ever runs. +type storageOp interface { + callback() chan error +} + type storageAdd struct { ctx context.Context convID chat1.ConversationID @@ -30,6 +36,8 @@ type storageAdd struct { cb chan error } +func (o storageAdd) callback() chan error { return o.cb } + type storageRemove struct { ctx context.Context convID chat1.ConversationID @@ -37,6 +45,8 @@ type storageRemove struct { cb chan error } +func (o storageRemove) callback() chan error { return o.cb } + type Indexer struct { globals.Contextified utils.DebugLabeler @@ -54,7 +64,7 @@ type Indexer struct { clock clockwork.Clock eg errgroup.Group uid gregor1.UID - storageCh chan any + storageCh chan storageOp maxSyncConvs int startSyncDelay time.Duration @@ -88,7 +98,7 @@ func NewIndexer(g *globals.Context) *Indexer { pokeSyncCh: make(chan struct{}, 100), clock: clockwork.NewRealClock(), flushDelay: 15 * time.Second, - storageCh: make(chan any, 100), + storageCh: make(chan storageOp, 100), stalledConvs: make(map[chat1.ConvIDStr]stalledConv), } switch idx.G().GetAppType() { @@ -313,9 +323,9 @@ func (idx *Indexer) Stop(ctx context.Context) chan struct{} { idx.Debug(ctx, "Stop: final flush failed: %s", err) } idx.store.ClearMemory() - // entries are keyed by convID only, and team convs are shared between - // users on this device: a count left by the previous user must not - // decide anything for the next one + // stall counts describe one indexer run, so drop them with it: a count + // carried into the next Start would clamp a conv on evidence gathered + // before the restart idx.stalledMu.Lock() idx.stalledConvs = make(map[chat1.ConvIDStr]stalledConv) idx.stalledMu.Unlock() @@ -400,8 +410,20 @@ func (idx *Indexer) consumeResultsForTest(convID chat1.ConversationID, err error // waiter - reindexConv does an unconditional receive - parked forever. In // SelectiveSync that leaks the sync's cancel func, so every later attempt sees // "already running" and background indexing is done until the process restarts. -func (idx *Indexer) storageDispatch(op any) bool { +// It also refuses to queue once the indexer is stopping. Nothing is left to run +// the op then, and the drain that would have released its callback has already +// walked the queue. +func (idx *Indexer) storageDispatch(op storageOp) bool { + idx.Lock() + stopCh := idx.stopCh + started := idx.started + idx.Unlock() + if !started { + return false + } select { + case <-stopCh: + return false case idx.storageCh <- op: return true default: @@ -432,13 +454,8 @@ func releaseStorageCB(cb chan error, err error) { func (idx *Indexer) drainStorageQueue() { for { select { - case iop := <-idx.storageCh: - switch op := iop.(type) { - case storageAdd: - releaseStorageCB(op.cb, errStorageStopped) - case storageRemove: - releaseStorageCB(op.cb, errStorageStopped) - } + case op := <-idx.storageCh: + releaseStorageCB(op.callback(), errStorageStopped) default: return } diff --git a/go/chat/search/storage.go b/go/chat/search/storage.go index d2f6ffc4b7a3..bef306fde19d 100644 --- a/go/chat/search/storage.go +++ b/go/chat/search/storage.go @@ -625,11 +625,14 @@ func (s *store) deleteTokenEntry(ctx context.Context, convID chat1.ConversationI s.tokenCache.Remove(cacheKey) - // Queue the delete rather than writing it through. Flush snapshots under the - // lock but writes outside it, so a delete applied straight to disk in that - // window was undone by the write that followed, and requeueing a failed - // write could put a deleted entry back. Ordering both through the pending - // set keeps the last operation the one that lands. + // Flush snapshots pending mutations under s.Lock, then applies them to disk + // after releasing the lock. A write-through delete could therefore race: + // + // flush snapshots old value -> delete removes disk key -> flush writes old value + // + // Queue a nil tombstone instead. A later flush applies it after any older + // snapshot, and requeue() will not replace the tombstone with a failed older + // write. This preserves the ordering of in-memory mutations. convIDStr := convID.ConvIDStr() if s.dirtyTokens[convIDStr] == nil { s.dirtyTokens[convIDStr] = make(map[string]*tokenEntry) @@ -1227,6 +1230,10 @@ func (s *store) requeue(tokens []tokenSnapshot, aliases map[string]*aliasEntry, if s.dirtyTokens[convIDStr] == nil { s.dirtyTokens[convIDStr] = make(map[string]*tokenEntry) } + // A value queued since the snapshot was taken - including a nil + // tombstone from a delete - describes a later state of the entry than + // the one whose write failed, so restoring the snapshot over it would + // roll that mutation back. if _, ok := s.dirtyTokens[convIDStr][snapshot.token]; ok { continue } diff --git a/go/chat/types/interfaces.go b/go/chat/types/interfaces.go index 77d86a2c95ae..67f0e4808629 100644 --- a/go/chat/types/interfaces.go +++ b/go/chat/types/interfaces.go @@ -642,6 +642,9 @@ type ParticipantSource interface { GetWithNotifyNonblock(ctx context.Context, uid gregor1.UID, convID chat1.ConversationID, dataSource InboxSourceDataSourceTyp) GetParticipantsFromUids(ctx context.Context, uids []gregor1.UID) ([]chat1.ConversationLocalParticipant, error) + // Invalidate drops any cached participant list for these convs, so the next + // read goes to the server rather than waiting out the cache freshness. + Invalidate(ctx context.Context, uid gregor1.UID, convIDs []chat1.ConversationID) } type EmojiSource interface { diff --git a/go/chat/types/types.go b/go/chat/types/types.go index 10d07040250c..7be8d8d3a425 100644 --- a/go/chat/types/types.go +++ b/go/chat/types/types.go @@ -874,6 +874,10 @@ func (d DummyParticipantSource) GetParticipantsFromUids( return nil, nil } +func (d DummyParticipantSource) Invalidate(ctx context.Context, uid gregor1.UID, + convIDs []chat1.ConversationID) { +} + type DummyEmojiSource struct{} var _ EmojiSource = (*DummyEmojiSource)(nil) diff --git a/go/teams/annotated_cache.go b/go/teams/annotated_cache.go index 5d4c63ee8f1f..c96c243d09cc 100644 --- a/go/teams/annotated_cache.go +++ b/go/teams/annotated_cache.go @@ -111,30 +111,20 @@ type annotatedTeamLoader func(ctx context.Context, g *libkb.GlobalContext, teamI // already-running load of the same team if there is one, and otherwise runs loader. func (c *annotatedTeamCache) load(mctx libkb.MetaContext, teamID keybase1.TeamID, loader annotatedTeamLoader) (res keybase1.AnnotatedTeam, err error) { for waits := 0; ; waits++ { - c.Lock() - if cached, ok := c.getLocked(mctx, teamID); ok { - c.Unlock() + cached, hit, wait, lead, startedGen := c.beginLoad(mctx, teamID, waits < maxAnnotatedTeamCacheWaits) + switch { + case hit: mctx.Debug("annotatedTeamCache: hit for %v", teamID) return cached, nil - } - if ch, ok := c.inflight[teamID]; ok && waits < maxAnnotatedTeamCacheWaits { - c.Unlock() + case wait != nil: mctx.Debug("annotatedTeamCache: waiting on in-flight load for %v", teamID) select { - case <-ch: + case <-wait: continue case <-mctx.Ctx().Done(): return res, mctx.Ctx().Err() } } - startedGen := c.gen[teamID] - ch := make(chan struct{}) - hasLead := false - if _, ok := c.inflight[teamID]; !ok { - c.inflight[teamID] = ch - hasLead = true - } - c.Unlock() // The leader must release its slot even if loader panics: the RPC // handler recovers, but a populated inflight entry whose channel is @@ -142,13 +132,8 @@ func (c *annotatedTeamCache) load(mctx libkb.MetaContext, teamID keybase1.TeamID // ctx dies, for the life of the process. Deferred rather than inline for // that reason; this path always returns below, so despite the enclosing // loop it is registered at most once. - if hasLead { - defer func() { - c.Lock() - delete(c.inflight, teamID) - c.Unlock() - close(ch) - }() + if lead != nil { + defer c.endLoad(teamID, lead) } // Age the entry from when the load started, not when it finished: the @@ -156,16 +141,59 @@ func (c *annotatedTeamCache) load(mctx libkb.MetaContext, teamID keybase1.TeamID // must not buy the result extra TTL it has already spent being stale. startedAt := mctx.G().Clock().Now() res, err = loader(mctx.Ctx(), mctx.G(), teamID) - - c.Lock() - if err == nil && c.gen[teamID] == startedGen { - c.entries[teamID] = annotatedTeamCacheEntry{team: res, cachedAt: startedAt} + if err == nil { + c.storeLoad(teamID, res, startedAt, startedGen) } - c.Unlock() return res, err } } +// beginLoad decides in one locked step what this caller does next: take a fresh +// cached team (hit), block on somebody else's load (wait), or run the load - as +// its leader if the slot was free (lead non-nil), otherwise alongside it. +// startedGen pins the invalidation generation the load reads at. +func (c *annotatedTeamCache) beginLoad(mctx libkb.MetaContext, teamID keybase1.TeamID, canWait bool) ( + cached keybase1.AnnotatedTeam, hit bool, wait, lead chan struct{}, startedGen uint64, +) { + c.Lock() + defer c.Unlock() + if cached, ok := c.getLocked(mctx, teamID); ok { + return cached, true, nil, nil, 0 + } + if ch, ok := c.inflight[teamID]; ok { + if canWait { + return cached, false, ch, nil, 0 + } + // Somebody is loading, but this caller has waited its share, so it goes + // ahead without claiming the slot. + return cached, false, nil, nil, c.gen[teamID] + } + ch := make(chan struct{}) + c.inflight[teamID] = ch + return cached, false, nil, ch, c.gen[teamID] +} + +// endLoad releases the leader's slot and wakes everyone waiting on it. +func (c *annotatedTeamCache) endLoad(teamID keybase1.TeamID, lead chan struct{}) { + c.Lock() + delete(c.inflight, teamID) + c.Unlock() + close(lead) +} + +// storeLoad memoizes a finished load, unless the team was invalidated while it +// ran - that result may have read pre-change data. +func (c *annotatedTeamCache) storeLoad(teamID keybase1.TeamID, team keybase1.AnnotatedTeam, + startedAt time.Time, startedGen uint64, +) { + c.Lock() + defer c.Unlock() + if c.gen[teamID] != startedGen { + return + } + c.entries[teamID] = annotatedTeamCacheEntry{team: team, cachedAt: startedAt} +} + func NewAnnotatedTeamCacheAndInstall(g *libkb.GlobalContext) { cache := newAnnotatedTeamCache() g.SetAnnotatedTeamCacher(cache)