gate RPCs and in-flight work on a chat session epoch

go/chat/globals/context.go

@@ -252,11 +252,17 @@ func ChatCtx(ctx context.Context, g *Context, mode keybase1.TLFIdentifyBehavior,
 	if _, ok := CtxTrace(res); !ok {
 		res = CtxAddLogTags(res, g)
 	}
+	if _, ok := ctxChatSession(res); !ok {
+		res = CtxStampChatSession(res, g)
+	}
 	return res
 }
 
 func BackgroundChatCtx(sourceCtx context.Context, g *Context) context.Context {
 	rctx := libkb.CopyTagsToBackground(sourceCtx)
+	if epoch, ok := ctxChatSession(sourceCtx); ok {
+		rctx = context.WithValue(rctx, chatSessionKey, epoch)
+	}
 
 	in := CtxIdentifyNotifier(sourceCtx)
 	if ident, breaks, ok := CtxIdentifyMode(sourceCtx); ok {

go/chat/globals/globals.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"regexp"
+	"sync"
 
 	"github.com/keybase/client/go/badges"
 	"github.com/keybase/client/go/chat/types"
@@ -52,6 +53,10 @@ type ChatContext struct {
 	EmojiSource          types.EmojiSource                // emoji support
 	EphemeralTracker     types.EphemeralTracker           // tracking of ephemeral msg caches
 	ArchiveRegistry      types.ChatArchiveRegistry        // Metadata store of chat archives
+
+	sessionMu      sync.Mutex
+	sessionEpoch   uint64
+	sessionBlocked bool
 }
 
 func (c *ChatContext) Describe() string {

go/chat/globals/session.go

@@ -0,0 +1,80 @@
+package globals
+
+import (
+	"context"
+
+	"github.com/keybase/client/go/libkb"
+)
+
+type (
+	chatSessionKeyTyp int
+)
+
+var chatSessionKey chatSessionKeyTyp
+
+func chatSessionNotReadyErr() error {
+	return libkb.NewLoginRequiredError("chat session not ready")
+}
+
+// Chat session gate: account switch is a logout barrier + login barrier.
+// Zero value is ready so unit tests that never login still work. The service
+// calls BeginChatLogout / MarkChatReady around stop/start of chat modules.
+func (c *ChatContext) setChatSession(blocked bool) {
+	c.sessionMu.Lock()
+	defer c.sessionMu.Unlock()
+	c.sessionEpoch++
+	c.sessionBlocked = blocked
+}
+
+func (c *ChatContext) BeginChatLogout() {
+	c.setChatSession(true)
+}
+
+func (c *ChatContext) MarkChatReady() {
+	c.setChatSession(false)
+}
+
+func (c *ChatContext) chatSessionSnapshot() (epoch uint64, ready bool) {
+	c.sessionMu.Lock()
+	defer c.sessionMu.Unlock()
+	return c.sessionEpoch, !c.sessionBlocked
+}
+
+func (c *ChatContext) ChatSessionReady() bool {
+	_, ready := c.chatSessionSnapshot()
+	return ready
+}
+
+func (c *ChatContext) AssertChatSessionReady() error {
+	if c.ChatSessionReady() {
+		return nil
+	}
+	return chatSessionNotReadyErr()
+}
+
+func CtxStampChatSession(ctx context.Context, g *Context) context.Context {
+	epoch, _ := g.chatSessionSnapshot()
+	return context.WithValue(ctx, chatSessionKey, epoch)
+}
+
+func BindChatSession(ctx context.Context, g *Context) (context.Context, error) {
+	epoch, ready := g.chatSessionSnapshot()
+	if !ready {
+		return ctx, chatSessionNotReadyErr()
+	}
+	return context.WithValue(ctx, chatSessionKey, epoch), nil
+}
+
+func ctxChatSession(ctx context.Context) (uint64, bool) {
+	epoch, ok := ctx.Value(chatSessionKey).(uint64)
+	return epoch, ok
+}
+
+func ChatSessionStale(ctx context.Context, g *Context) bool {
+	stamp, ok := ctxChatSession(ctx)
+	if !ok {
+		return !g.ChatSessionReady()
+	}
+	epoch, ready := g.chatSessionSnapshot()
+	return !ready || stamp != epoch
+}

go/chat/globals/session_test.go

@@ -0,0 +1,35 @@
+package globals
+
+import (
+	"context"
+	"testing"
+
+	"github.com/keybase/client/go/libkb"
+	"github.com/stretchr/testify/require"
+)
+
+func TestChatSessionGate(t *testing.T) {
+	c := &ChatContext{}
+	g := &Context{ChatContext: c}
+	require.True(t, c.ChatSessionReady())
+	require.NoError(t, c.AssertChatSessionReady())
+
+	c.BeginChatLogout()
+	require.False(t, c.ChatSessionReady())
+	err := c.AssertChatSessionReady()
+	require.Error(t, err)
+	_, ok := err.(libkb.LoginRequiredError)
+	require.True(t, ok)
+	_, err = BindChatSession(context.Background(), g)
+	require.Error(t, err)
+
+	ctx := CtxStampChatSession(context.Background(), g)
+	require.True(t, ChatSessionStale(ctx, g))
+
+	c.MarkChatReady()
+	require.True(t, c.ChatSessionReady())
+	ctx2, err := BindChatSession(context.Background(), g)
+	require.NoError(t, err)
+	require.False(t, ChatSessionStale(ctx2, g))
+	require.True(t, ChatSessionStale(ctx, g), "old epoch must be stale after login")
+}

go/chat/localizer.go

@@ -548,6 +548,9 @@ func (s *localizerPipeline) jobPulled(ctx context.Context, job *localizerPipelin
 func (s *localizerPipeline) localizeConversations(localizeJob *localizerPipelineJob) (err error) {
 	ctx := localizeJob.ctx
 	uid := localizeJob.uid
+	if globals.ChatSessionStale(ctx, s.G()) {
+		return storage.NewAbortedError()
+	}
 	defer s.Trace(ctx, &err, "localizeConversations")()
 
 	// Fetch conversation local information in parallel

go/chat/retry.go

@@ -299,7 +299,7 @@ func (f *FetchRetrier) Failure(ctx context.Context, uid gregor1.UID, desc types.
 	defer f.Trace(ctx, nil, "Failure(%s)", desc)()
 	f.Lock()
 	defer f.Unlock()
-	if !f.running {
+	if !f.running || !f.G().ChatSessionReady() {
 		f.Debug(ctx, "Failure: not starting new retrier, not running")
 		return
 	}

go/chat/session_rpc_test.go

@@ -0,0 +1,35 @@
+package chat
+
+import (
+	"context"
+	"testing"
+
+	"github.com/keybase/client/go/chat/storage"
+	"github.com/keybase/client/go/protocol/chat1"
+	"github.com/keybase/client/go/protocol/gregor1"
+	"github.com/stretchr/testify/require"
+)
+
+func TestInboxLoaderRejectsUntilReady(t *testing.T) {
+	ctc := makeChatTestContext(t, "TestInboxLoaderRejectsUntilReady", 1)
+	defer ctc.cleanup()
+	users := ctc.users()
+	h := ctc.as(t, users[0]).h
+	tc := ctc.world.Tcs[users[0].Username]
+	uid := gregor1.UID(users[0].GetUID().ToBytes())
+	loader := NewUIInboxLoader(tc.Context())
+	tc.ChatG.UIInboxLoader = loader
+	loader.Start(context.Background(), uid)
+	defer func() { <-loader.Stop(context.Background()) }()
+
+	// Unstamped ctx: stale iff ChatSessionReady is false, independent of setup stamps.
+	ctx := context.Background()
+	h.G().BeginChatLogout()
+	err := loader.UpdateConvs(ctx, []chat1.ConversationID{{0x01}})
+	require.Error(t, err)
+	require.ErrorAs(t, err, new(storage.AbortedError))
+
+	h.G().MarkChatReady()
+	_, err = loader.sessionUID(ctx)
+	require.NoError(t, err)
+}

go/chat/storage/inbox.go

@@ -202,6 +202,9 @@ func (i *Inbox) readDiskVersions(ctx context.Context, uid gregor1.UID, useInMemo
 	if err := isAbortedRequest(ctx); err != nil {
 		return ibox, err
 	}
+	if globals.ChatSessionStale(ctx, i.G()) {
+		return ibox, NewAbortedError()
+	}
 	if err := i.missIfWrongSessionUID(uid); err != nil {
 		return ibox, err
 	}
@@ -272,6 +275,9 @@ func (i *Inbox) readDiskIndex(ctx context.Context, uid gregor1.UID, useInMemory
 	if err := isAbortedRequest(ctx); err != nil {
 		return ibox, err
 	}
+	if globals.ChatSessionStale(ctx, i.G()) {
+		return ibox, NewAbortedError()
+	}
 	if err := i.missIfWrongSessionUID(uid); err != nil {
 		return ibox, err
 	}