gotd: add extra cache for left channels
This commit is contained in:
@@ -7,6 +7,7 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/go-faster/errors"
|
"github.com/go-faster/errors"
|
||||||
|
"go.mau.fi/util/exsync"
|
||||||
"go.opentelemetry.io/otel/trace"
|
"go.opentelemetry.io/otel/trace"
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
"golang.org/x/sync/errgroup"
|
"golang.org/x/sync/errgroup"
|
||||||
@@ -45,8 +46,9 @@ type internalState struct {
|
|||||||
idleTimeout *time.Timer
|
idleTimeout *time.Timer
|
||||||
|
|
||||||
// Channel states.
|
// Channel states.
|
||||||
channels map[int64]*channelState
|
channels map[int64]*channelState
|
||||||
channelsLock sync.Mutex
|
channelsLock sync.Mutex
|
||||||
|
recentlyLeftChannels *exsync.Set[int64]
|
||||||
|
|
||||||
// Immutable fields.
|
// Immutable fields.
|
||||||
client API
|
client API
|
||||||
@@ -84,7 +86,8 @@ func newState(ctx context.Context, cfg stateConfig) *internalState {
|
|||||||
date: cfg.State.Date,
|
date: cfg.State.Date,
|
||||||
idleTimeout: time.NewTimer(idleTimeout),
|
idleTimeout: time.NewTimer(idleTimeout),
|
||||||
|
|
||||||
channels: make(map[int64]*channelState),
|
channels: make(map[int64]*channelState),
|
||||||
|
recentlyLeftChannels: exsync.NewSet[int64](),
|
||||||
|
|
||||||
client: cfg.RawClient,
|
client: cfg.RawClient,
|
||||||
log: cfg.Logger,
|
log: cfg.Logger,
|
||||||
@@ -341,14 +344,17 @@ func (s *internalState) handleChannel(ctx context.Context, channelID int64, date
|
|||||||
s.log.Error("Pts validation failed", zap.Error(err), zap.Any("update", cu.update))
|
s.log.Error("Pts validation failed", zap.Error(err), zap.Any("update", cu.update))
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
found := false
|
||||||
for _, ent := range cu.entities.Chats {
|
for _, ent := range cu.entities.Chats {
|
||||||
if ent.GetID() == channelID {
|
if ent.GetID() == channelID {
|
||||||
|
found = true
|
||||||
switch te := ent.(type) {
|
switch te := ent.(type) {
|
||||||
case *tg.Channel:
|
case *tg.Channel:
|
||||||
if te.Left {
|
if te.Left {
|
||||||
s.log.Info("Not adding new channel state for left channel", zap.Int64("channel_id", channelID))
|
s.log.Info("Not adding new channel state for left channel", zap.Int64("channel_id", channelID))
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
s.recentlyLeftChannels.Remove(channelID)
|
||||||
case *tg.ChannelForbidden:
|
case *tg.ChannelForbidden:
|
||||||
s.log.Info("Not adding new channel state for forbidden channel", zap.Int64("channel_id", channelID))
|
s.log.Info("Not adding new channel state for forbidden channel", zap.Int64("channel_id", channelID))
|
||||||
return nil
|
return nil
|
||||||
@@ -356,6 +362,10 @@ func (s *internalState) handleChannel(ctx context.Context, channelID int64, date
|
|||||||
break
|
break
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if !found && s.recentlyLeftChannels.Has(channelID) {
|
||||||
|
s.log.Info("Not adding new channel state for recently left channel", zap.Int64("channel_id", channelID))
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
s.channelsLock.Lock()
|
s.channelsLock.Lock()
|
||||||
state, ok := s.channels[channelID]
|
state, ok := s.channels[channelID]
|
||||||
|
|||||||
@@ -51,9 +51,11 @@ func (s *internalState) applyCombined(ctx context.Context, comb *tg.UpdatesCombi
|
|||||||
ptsChanged = true
|
ptsChanged = true
|
||||||
continue
|
continue
|
||||||
case *tg.UpdateChannelTooLong:
|
case *tg.UpdateChannelTooLong:
|
||||||
|
s.channelsLock.Lock()
|
||||||
st, ok := s.channels[u.ChannelID]
|
st, ok := s.channels[u.ChannelID]
|
||||||
|
s.channelsLock.Unlock()
|
||||||
if !ok {
|
if !ok {
|
||||||
s.log.Debug("ChannelTooLong for channel that is not in the internalState, update ignored", zap.Int64("channel_id", u.ChannelID))
|
s.log.Info("ChannelTooLong for channel that is not in the internalState, update ignored", zap.Int64("channel_id", u.ChannelID))
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
if err := st.Push(ctx, channelUpdate{
|
if err := st.Push(ctx, channelUpdate{
|
||||||
|
|||||||
Reference in New Issue
Block a user