<commit_before>package chat

import (
	"bytes"
	"encoding/hex"
	"errors"
	"fmt"
	"sync"
	"time"

	"strings"

	"github.com/keybase/client/go/badges"
	"github.com/keybase/client/go/chat/globals"
	"github.com/keybase/client/go/chat/pager"
	"github.com/keybase/client/go/chat/storage"
	"github.com/keybase/client/go/chat/types"
	"github.com/keybase/client/go/chat/utils"
	"github.com/keybase/client/go/gregor"
	"github.com/keybase/client/go/libkb"
	"github.com/keybase/client/go/protocol/chat1"
	"github.com/keybase/client/go/protocol/gregor1"
	"github.com/keybase/client/go/protocol/keybase1"
	"github.com/keybase/client/go/teams"
	"github.com/keybase/clockwork"
	"github.com/keybase/go-codec/codec"
	"golang.org/x/net/context"
)

var errPushOrdererMissingLatestInboxVersion = errors.New("no latest inbox version")

type messageWaiterEntry struct {
	vers chat1.InboxVers
	cb   chan struct{}
}

type gregorMessageOrderer struct {
	globals.Contextified
	utils.DebugLabeler
	sync.Mutex

	clock      clockwork.Clock
	waiters    map[string][]messageWaiterEntry
	noInboxDur time.Duration
}

func newGregorMessageOrderer(g *globals.Context) *gregorMessageOrderer {
	return &gregorMessageOrderer{
		Contextified: globals.NewContextified(g),
		DebugLabeler: utils.NewDebugLabeler(g.GetLog(), "gregorMessageOrderer", false),
		waiters:      make(map[string][]messageWaiterEntry),
		clock:        clockwork.NewRealClock(),
		noInboxDur:   time.Second,
	}
}

func (g *gregorMessageOrderer) msgKey(uid gregor1.UID, vers chat1.InboxVers) string {
	return fmt.Sprintf("%s:%d", uid, vers)
}

func (g *gregorMessageOrderer) isUIDKey(key string, uid gregor1.UID) bool {
	toks := strings.Split(key, ":")
	return toks[0] == uid.String()
}

func (g *gregorMessageOrderer) latestInboxVersion(ctx context.Context, uid gregor1.UID) (chat1.InboxVers, error) {
	ibox := storage.NewInbox(g.G())
	vers, err := ibox.Version(ctx, uid)
	if err != nil {
		return 0, err
	}
	if vers == 0 {
		return 0, errPushOrdererMissingLatestInboxVersion
	}
	return vers, nil
}

func (g *gregorMessageOrderer) addToWaitersLocked(ctx context.Context, uid gregor1.UID, storedVers,
	msgVers chat1.InboxVers) (res []messageWaiterEntry) {
	for i := storedVers + 1; i < msgVers; i++ {
		entry := messageWaiterEntry{
			vers: msgVers,
			cb:   make(chan struct{}),
		}
		res = append(res, entry)
		key := g.msgKey(uid, i)
		g.waiters[key] = append(g.waiters[key], entry)
	}
	return res
}

func (g *gregorMessageOrderer) waitOnWaiters(ctx context.Context, vers chat1.InboxVers,
	waiters []messageWaiterEntry) (res chan struct{}) {
	res = make(chan struct{})
	go func(ctx context.Context) {
		for _, w := range waiters {
			select {
			case <-w.cb:
			case <-ctx.Done():
				g.Debug(ctx, "waitOnWaiters: context cancelled: waiter: %d target: %d", vers, w.vers)
			}
		}
		close(res)
	}(globals.BackgroundChatCtx(ctx, g.G()))
	return res
}

func (g *gregorMessageOrderer) cleanupAfterTimeoutLocked(uid gregor1.UID, vers chat1.InboxVers) {
	for k, v := range g.waiters {
		if g.isUIDKey(k, uid) {
			var newv []messageWaiterEntry
			for _, w := range v {
				if w.vers != vers {
					newv = append(newv, w)
				}
			}
			if len(newv) == 0 {
				delete(g.waiters, k)
			} else {
				g.waiters[k] = newv
			}
		}
	}
}

func (g *gregorMessageOrderer) WaitForTurn(ctx context.Context, uid gregor1.UID,
	newVers chat1.InboxVers) (res chan struct{}) {
	res = make(chan struct{})
	// Out of order update, we are going to wait a fixed amount of time for the correctly
	// ordered update
	deadline := g.clock.Now().Add(time.Second)
	go func(ctx context.Context) {
		defer close(res)
		g.Lock()
		var dur time.Duration
		vers, err := g.latestInboxVersion(ctx, uid)
		if err != nil {
			if err == errPushOrdererMissingLatestInboxVersion {
				g.Debug(ctx, "WaitForTurn: no inbox version, setting fixed duration of %v", g.noInboxDur)
			} else {
				vers = newVers - 1
				g.Debug(ctx, "WaitForTurn: failed to get current inbox version: %v. Proceeding with vers %d",
					err, vers)
			}
			dur = g.noInboxDur
		} else {
			// add extra time if we are multiple updates behind
			dur = time.Duration(newVers-vers-1) * time.Second
			if dur < 0 {
				dur = 0
			}
			// cap at a minute
			if dur > time.Minute {
				dur = time.Minute
			}
		}
		deadline = deadline.Add(dur)
		waiters := g.addToWaitersLocked(ctx, uid, vers, newVers)
		g.Unlock()
		if len(waiters) == 0 {
			return
		}
		waitBegin := g.clock.Now()
		g.Debug(ctx,
			"WaitForTurn: out of order update received, waiting on %d updates: vers: %d newVers: %d dur: %v",
			len(waiters), vers, newVers, dur+time.Second)
		ctx, cancel := context.WithCancel(ctx)
		defer cancel()
		select {
		case <-g.waitOnWaiters(ctx, newVers, waiters):
			g.Debug(ctx, "WaitForTurn: cleared by earlier messages: vers: %d wait: %v", newVers,
				g.clock.Now().Sub(waitBegin))
		case <-g.clock.AfterTime(deadline):
			g.Debug(ctx, "WaitForTurn: timeout reached, charging forward: vers: %d wait: %v", newVers,
				g.clock.Now().Sub(waitBegin))
			g.Lock()
			g.cleanupAfterTimeoutLocked(uid, newVers)
			g.Unlock()
		}<commit_msg>more clear handling of inbox error in orderer (#18297)<commit_after>package chat

import (
	"bytes"
	"encoding/hex"
	"errors"
	"fmt"
	"sync"
	"time"

	"strings"

	"github.com/keybase/client/go/badges"
	"github.com/keybase/client/go/chat/globals"
	"github.com/keybase/client/go/chat/pager"
	"github.com/keybase/client/go/chat/storage"
	"github.com/keybase/client/go/chat/types"
	"github.com/keybase/client/go/chat/utils"
	"github.com/keybase/client/go/gregor"
	"github.com/keybase/client/go/libkb"
	"github.com/keybase/client/go/protocol/chat1"
	"github.com/keybase/client/go/protocol/gregor1"
	"github.com/keybase/client/go/protocol/keybase1"
	"github.com/keybase/client/go/teams"
	"github.com/keybase/clockwork"
	"github.com/keybase/go-codec/codec"
	"golang.org/x/net/context"
)

var errPushOrdererMissingLatestInboxVersion = errors.New("no latest inbox version")

type messageWaiterEntry struct {
	vers chat1.InboxVers
	cb   chan struct{}
}

type gregorMessageOrderer struct {
	globals.Contextified
	utils.DebugLabeler
	sync.Mutex

	clock   clockwork.Clock
	waiters map[string][]messageWaiterEntry
}

func newGregorMessageOrderer(g *globals.Context) *gregorMessageOrderer {
	return &gregorMessageOrderer{
		Contextified: globals.NewContextified(g),
		DebugLabeler: utils.NewDebugLabeler(g.GetLog(), "gregorMessageOrderer", false),
		waiters:      make(map[string][]messageWaiterEntry),
		clock:        clockwork.NewRealClock(),
	}
}

func (g *gregorMessageOrderer) msgKey(uid gregor1.UID, vers chat1.InboxVers) string {
	return fmt.Sprintf("%s:%d", uid, vers)
}

func (g *gregorMessageOrderer) isUIDKey(key string, uid gregor1.UID) bool {
	toks := strings.Split(key, ":")
	return toks[0] == uid.String()
}

func (g *gregorMessageOrderer) latestInboxVersion(ctx context.Context, uid gregor1.UID) (chat1.InboxVers, error) {
	ibox := storage.NewInbox(g.G())
	vers, err := ibox.Version(ctx, uid)
	if err != nil {
		return 0, err
	}
	if vers == 0 {
		return 0, errPushOrdererMissingLatestInboxVersion
	}
	return vers, nil
}

func (g *gregorMessageOrderer) addToWaitersLocked(ctx context.Context, uid gregor1.UID, storedVers,
	msgVers chat1.InboxVers) (res []messageWaiterEntry) {
	for i := storedVers + 1; i < msgVers; i++ {
		entry := messageWaiterEntry{
			vers: msgVers,
			cb:   make(chan struct{}),
		}
		res = append(res, entry)
		key := g.msgKey(uid, i)
		g.waiters[key] = append(g.waiters[key], entry)
	}
	return res
}

func (g *gregorMessageOrderer) waitOnWaiters(ctx context.Context, vers chat1.InboxVers,
	waiters []messageWaiterEntry) (res chan struct{}) {
	res = make(chan struct{})
	go func(ctx context.Context) {
		for _, w := range waiters {
			select {
			case <-w.cb:
			case <-ctx.Done():
				g.Debug(ctx, "waitOnWaiters: context cancelled: waiter: %d target: %d", vers, w.vers)
			}
		}
		close(res)
	}(globals.BackgroundChatCtx(ctx, g.G()))
	return res
}

func (g *gregorMessageOrderer) cleanupAfterTimeoutLocked(uid gregor1.UID, vers chat1.InboxVers) {
	for k, v := range g.waiters {
		if g.isUIDKey(k, uid) {
			var newv []messageWaiterEntry
			for _, w := range v {
				if w.vers != vers {
					newv = append(newv, w)
				}
			}
			if len(newv) == 0 {
				delete(g.waiters, k)
			} else {
				g.waiters[k] = newv
			}
		}
	}
}

func (g *gregorMessageOrderer) WaitForTurn(ctx context.Context, uid gregor1.UID,
	newVers chat1.InboxVers) (res chan struct{}) {
	res = make(chan struct{})
	// Out of order update, we are going to wait a fixed amount of time for the correctly
	// ordered update
	deadline := g.clock.Now().Add(time.Second)
	go func(ctx context.Context) {
		defer close(res)
		g.Lock()
		var dur time.Duration
		vers, err := g.latestInboxVersion(ctx, uid)
		if err != nil {
			if newVers >= 2 {
				// If we fail to get the inbox version, then the general goal is to just simulate like
				// we are looking for the previous update to the one we just got (newVers). In order to do
				// this, we act like our previous inbox version was just missing the inbox version one less
				// than newVers. That means we are waiting for this single update (which probably isn't
				// coming, we usually get here if we miss the inbox cache on disk).
				vers = newVers - 2
			} else {
				vers = 0
			}
			g.Debug(ctx, "WaitForTurn: failed to get current inbox version: %s, proceeding with vers %d",
				err, vers)
		}
		// add extra time if we are multiple updates behind
		dur = time.Duration(newVers-vers-1) * time.Second
		if dur < 0 {
			dur = 0
		}
		// cap at a minute
		if dur > time.Minute {
			dur = time.Minute
		}

		deadline = deadline.Add(dur)
		waiters := g.addToWaitersLocked(ctx, uid, vers, newVers)
		g.Unlock()
		if len(waiters) == 0 {
			return
		}
		waitBegin := g.clock.Now()
		g.Debug(ctx,
			"WaitForTurn: out of order update received, waiting on %d updates: vers: %d newVers: %d dur: %v",
			len(waiters), vers, newVers, dur+time.Second)
		ctx, cancel := context.WithCancel(ctx)
		defer cancel()
		select {
		case <-g.waitOnWaiters(ctx, newVers, waiters):
			g.Debug(ctx, "WaitForTurn: cleared by earlier messages: vers: %d wait: %v", newVers,
				g.clock.Now().Sub(waitBegin))
		case <-g.clock.AfterTime(deadline):
			g.Debug(ctx, "WaitForTurn: timeout reached, charging forward: vers: %d wait: %v", newVers,
				g.clock.Now().Sub(waitBegin))
			g.Lock()
			g.cleanupAfterTimeoutLocked(uid, newVers)
			g.Unlock()
		}
{'fraction_non_alphanumeric': 0.1355381380233225, 'fraction_numerical': 0.005830625119480023, 'mean_word_length': 3.819438046982957, 'pattern_counts': {'":': 2, '<': 15, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '755236', 'n_tokens_mistral': 4193, 'n_tokens_neox': 3847, 'n_words': 926}