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..6b7bd9259134 100644 --- a/go/chat/participantsource.go +++ b/go/chat/participantsource.go @@ -18,10 +18,21 @@ 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. +// +// 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 { globals.Contextified utils.DebugLabeler @@ -90,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) { @@ -130,6 +156,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 +177,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/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 new file mode 100644 index 000000000000..0a7d04af1bf4 --- /dev/null +++ b/go/chat/search/flush_test.go @@ -0,0 +1,648 @@ +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 := chat1.MessageID(0); i < maxDirtyEntries-1; i++ { + te := newTokenEntry() + te.MsgIDs[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 := chat1.MessageID(0); i < maxDirtyEntries*2; i++ { + te := newTokenEntry() + te.MsgIDs[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") +} + +// 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{ + 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..338d6c08b926 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" @@ -22,20 +23,30 @@ 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 msgs []chat1.MessageUnboxed - cb chan struct{} + cb chan error } +func (o storageAdd) callback() chan error { return o.cb } + type storageRemove struct { ctx context.Context convID chat1.ConversationID msgs []chat1.MessageUnboxed - cb chan struct{} + cb chan error } +func (o storageRemove) callback() chan error { return o.cb } + type Indexer struct { globals.Contextified utils.DebugLabeler @@ -53,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 @@ -62,6 +73,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 @@ -81,7 +98,8 @@ 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() { case libkb.MobileAppType: @@ -297,7 +315,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() + // 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() idx.started = false close(idx.stopCh) go func() { @@ -373,11 +404,61 @@ 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. +// 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: 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) +} + +// 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 op := <-idx.storageCh: + releaseStorageCB(op.callback(), errStorageStopped) + default: + return + } } } @@ -388,6 +469,7 @@ func (idx *Indexer) storageLoop(stopCh chan struct{}) error { select { case <-stopCh: idx.Debug(ctx, "storageLoop: shutting down") + idx.drainStorageQueue() return nil case iop := <-idx.storageCh: switch op := iop.(type) { @@ -397,14 +479,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 +500,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 +541,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 +559,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 +587,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 +605,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 +642,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 +830,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 +945,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 +1086,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 +1101,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..37d7ef796501 --- /dev/null +++ b/go/chat/search/stall_test.go @@ -0,0 +1,242 @@ +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") + cb := make(chan error, 1) + idx.storageCh <- storageAdd{cb: cb} + + // 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: + 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..bef306fde19d 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,33 @@ func (s *store) deleteTokenEntry(ctx context.Context, convID chat1.ConversationI s.tokenCache.Remove(cacheKey) + // 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 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 +810,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 +852,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 +891,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 +907,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 +983,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 +1014,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 +1023,8 @@ func (s *store) Add(ctx context.Context, convID chat1.ConversationID, return err } } + seenIDs[msg.GetMessageID()] = chat1.EmptyStruct{} + modified = true } return nil } @@ -860,14 +1050,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 +1075,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 +1109,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 +1128,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 +1162,96 @@ 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) + } + // 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 + } + 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++ + } +} 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/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/globals.go b/go/libkb/globals.go index 8fb1f0814343..f65272313af5 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) @@ -753,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 } @@ -765,9 +766,21 @@ func (g *GlobalContext) GetImplicitTeamCacher() MemLRUer { } func (g *GlobalContext) SetImplicitTeamCacher(l MemLRUer) { + g.cacheMu.Lock() + defer g.cacheMu.Unlock() + g.iteamCacher = l +} + +func (g *GlobalContext) GetAnnotatedTeamCacher() AnnotatedTeamCacher { g.cacheMu.RLock() defer g.cacheMu.RUnlock() - g.iteamCacher = l + return g.annotatedTeamCacher +} + +func (g *GlobalContext) SetAnnotatedTeamCacher(c AnnotatedTeamCacher) { + g.cacheMu.Lock() + defer g.cacheMu.Unlock() + g.annotatedTeamCacher = c } func (g *GlobalContext) GetKVRevisionCache() KVRevisionCacher { @@ -777,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/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/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/libkb/proof_cache.go b/go/libkb/proof_cache.go index 0a75b516e281..702492bbf6e3 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,104 @@ 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) +} + +// 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) { + 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) { + // 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{})} + pc.flights.Add(key, f) + return f, nil } func NewProofCache(g *GlobalContext, capac int) *ProofCache { @@ -124,6 +223,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..84d2405f409a --- /dev/null +++ b/go/libkb/proof_cache_flight_test.go @@ -0,0 +1,89 @@ +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 TestProofCheckFlightUnusableResultDropped(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) + + // 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 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) + } +} diff --git a/go/teams/annotated_cache.go b/go/teams/annotated_cache.go new file mode 100644 index 000000000000..c96c243d09cc --- /dev/null +++ b/go/teams/annotated_cache.go @@ -0,0 +1,210 @@ +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++ { + cached, hit, wait, lead, startedGen := c.beginLoad(mctx, teamID, waits < maxAnnotatedTeamCacheWaits) + switch { + case hit: + mctx.Debug("annotatedTeamCache: hit for %v", teamID) + return cached, nil + case wait != nil: + mctx.Debug("annotatedTeamCache: waiting on in-flight load for %v", teamID) + select { + case <-wait: + continue + case <-mctx.Ctx().Done(): + return res, mctx.Ctx().Err() + } + } + + // 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 lead != nil { + defer c.endLoad(teamID, lead) + } + + // 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) + if err == nil { + c.storeLoad(teamID, res, startedAt, startedGen) + } + 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) + 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() 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 }