diff --git a/go.mod b/go.mod index 440165b..c17ee2d 100644 --- a/go.mod +++ b/go.mod @@ -8,6 +8,7 @@ require ( ) require ( + github.com/google/uuid v1.6.0 // indirect github.com/mattn/go-colorable v0.1.14 // indirect github.com/mattn/go-isatty v0.0.20 // indirect golang.org/x/sys v0.29.0 // indirect diff --git a/go.sum b/go.sum index 225abaf..e8d7e57 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE= github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8= github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= diff --git a/v1/core/dispatcher.go b/v1/core/dispatcher.go index 34401c9..de5833e 100644 --- a/v1/core/dispatcher.go +++ b/v1/core/dispatcher.go @@ -4,8 +4,10 @@ import ( "context" "errors" mcontext "github.com/abelith/gomax/v1/context" + "github.com/abelith/gomax/v1/isolation" "github.com/abelith/gomax/v1/state" "github.com/abelith/gomax/v1/state/memory" + "github.com/google/uuid" maxbot "github.com/max-messenger/max-bot-api-client-go/v2" "github.com/max-messenger/max-bot-api-client-go/v2/model" "github.com/rs/zerolog/log" @@ -14,12 +16,14 @@ import ( ) const ( - defaultTTL = 5 * time.Second + defaultTTL = 30 * time.Second + defaultLockTimeout = 10 * time.Second + defaultHandlerTimeout = 10 * time.Second ) type EventIsolation interface { - Lock(ctx context.Context, userID, chatID int) error - Unlock(ctx context.Context, userID, chatID int) error + Lock(ctx context.Context, key isolation.Key) error + Unlock(ctx context.Context, key isolation.Key) error } type FSMStorage interface { @@ -78,8 +82,8 @@ func NewDispatcher(api *maxbot.Api, options ...DispatcherOption) *Dispatcher { return d } -func (d *Dispatcher) Handle(handler UpdateHandler, filters func(c *mcontext.Context) bool) { - d.rt.Handle(handler, filters) +func (d *Dispatcher) Handle(handler UpdateHandler, filters ...func(c *mcontext.Context) bool) { + d.rt.Handle(handler, filters...) } func (d *Dispatcher) HandleFunc(handler func(c *mcontext.Context) error, filters ...func(c *mcontext.Context) bool) { @@ -101,7 +105,6 @@ func (d *Dispatcher) Start(ctx context.Context) error { } d.handleUpdates(ch) - <-ctx.Done() d.wg.Wait() return nil } @@ -109,21 +112,44 @@ func (d *Dispatcher) Start(ctx context.Context) error { func (d *Dispatcher) handleUpdates(ch <-chan model.Update) { for upd := range ch { d.wg.Go(func() { - ctx := context.Background() + key := isolation.Key{ + UpdateID: uuid.New(), + UserID: upd.UserID, + ChatID: upd.ChatID, + } + l := log.With(). + Str("update_id", key.UpdateID.String()). + Int64("user_id", key.UserID). + Int64("chat_id", key.ChatID). + Logger() - _ = d.eventIsolation.Lock(ctx, int(upd.UserID), int(upd.ChatID)) // todo: handle error somehow (retry or smth) + lockCtx, cancel := context.WithTimeout(context.Background(), defaultLockTimeout) + defer cancel() + + if err := d.eventIsolation.Lock(lockCtx, key); err != nil { + log.Error(). + Err(err). + Int64("user_id", key.UserID). + Int64("chat_id", key.ChatID). + Msg("failed to lock on update") + return + } defer func() { - err := d.eventIsolation.Unlock(ctx, int(upd.UserID), int(upd.ChatID)) - if err != nil { - log.Warn(). - Err(err). - Int64("user_id", upd.UserID). - Int64("chat_id", upd.ChatID). - Msg("failed to unlock on update") + unlockCtx, unlockCancel := context.WithTimeout(context.Background(), defaultLockTimeout) + defer unlockCancel() + + err := d.eventIsolation.Unlock(unlockCtx, key) + switch { + case errors.Is(err, isolation.ErrNotOwner): + l.Warn().Msg("lock expired before handler finished") + case err != nil: + l.Warn().Err(err).Msg("failed to unlock on update") } }() - d.handleUpdate(ctx, upd) + handleCtx, cancel := context.WithTimeout(context.Background(), defaultHandlerTimeout) + defer cancel() + d.handleUpdate(handleCtx, upd) }) } } diff --git a/v1/core/pollers.go b/v1/core/pollers.go index f1e253c..fed31fd 100644 --- a/v1/core/pollers.go +++ b/v1/core/pollers.go @@ -128,6 +128,7 @@ func (w *Webhook) Start(_ context.Context) (<-chan model.Update, error) { } func (w *Webhook) Stop(ctx context.Context) error { + close(w.ch) return w.srv.Shutdown(ctx) } @@ -176,8 +177,6 @@ func (lp *LongPoller) Start(ctx context.Context) (<-chan model.Update, error) { for _, update := range updates { select { - case <-ctx.Done(): - return case ch <- update: } } diff --git a/v1/isolation/isolation.go b/v1/isolation/isolation.go new file mode 100644 index 0000000..e66bea4 --- /dev/null +++ b/v1/isolation/isolation.go @@ -0,0 +1,22 @@ +package isolation + +import ( + "errors" + + "github.com/google/uuid" +) + +var ErrNotOwner = errors.New("isolation: lock is not held by this owner") + +type Key struct { + UpdateID uuid.UUID // владелец + UserID int64 + ChatID int64 +} + +type Scope struct { + UserID int64 + ChatID int64 +} + +func (k Key) Scope() Scope { return Scope{UserID: k.UserID, ChatID: k.ChatID} } diff --git a/v1/state/memory/memory.go b/v1/state/memory/memory.go index c891c00..22ab55f 100644 --- a/v1/state/memory/memory.go +++ b/v1/state/memory/memory.go @@ -2,7 +2,9 @@ package memory import ( "context" + "github.com/abelith/gomax/v1/isolation" "github.com/abelith/gomax/v1/state" + "github.com/google/uuid" "sync" "time" ) @@ -13,7 +15,7 @@ type FSMStorage struct { type EventIsolation struct { mu sync.Mutex - locks map[eventIsolationKey]*lockEntry + locks map[isolation.Scope]*lockEntry ttl time.Duration } @@ -23,6 +25,7 @@ type eventIsolationKey struct { } type lockEntry struct { + owner uuid.UUID timer *time.Timer done chan struct{} } @@ -52,19 +55,19 @@ func (f *FSMStorage) Delete(_ context.Context, userID int) error { func NewEventIsolation(ttl time.Duration) *EventIsolation { return &EventIsolation{ - locks: make(map[eventIsolationKey]*lockEntry), + locks: make(map[isolation.Scope]*lockEntry), ttl: ttl, } } -func (ei *EventIsolation) Lock(ctx context.Context, userID, chatID int) error { - key := eventIsolationKey{userID: userID, chatID: chatID} +func (ei *EventIsolation) Lock(ctx context.Context, key isolation.Key) error { + scope := key.Scope() for { ei.mu.Lock() - entry, busy := ei.locks[key] + entry, busy := ei.locks[scope] if !busy { - ei.acquireLocked(key) + ei.acquireLocked(scope, key.UpdateID) ei.mu.Unlock() return nil } @@ -79,32 +82,34 @@ func (ei *EventIsolation) Lock(ctx context.Context, userID, chatID int) error { } } -func (ei *EventIsolation) Unlock(_ context.Context, userID, chatID int) error { - key := eventIsolationKey{userID: userID, chatID: chatID} +func (ei *EventIsolation) Unlock(_ context.Context, key isolation.Key) error { + scope := key.Scope() ei.mu.Lock() defer ei.mu.Unlock() - if entry, ok := ei.locks[key]; ok { - ei.releaseLocked(key, entry) + entry, ok := ei.locks[scope] + if !ok || entry.owner != key.UpdateID { + return isolation.ErrNotOwner } + ei.releaseLocked(scope, entry) return nil } -func (ei *EventIsolation) acquireLocked(key eventIsolationKey) { - entry := &lockEntry{done: make(chan struct{})} +func (ei *EventIsolation) acquireLocked(scope isolation.Scope, owner uuid.UUID) { + entry := &lockEntry{owner: owner, done: make(chan struct{})} entry.timer = time.AfterFunc(ei.ttl, func() { ei.mu.Lock() defer ei.mu.Unlock() - if cur, ok := ei.locks[key]; ok && cur == entry { - ei.releaseLocked(key, entry) + if cur, ok := ei.locks[scope]; ok && cur == entry { + ei.releaseLocked(scope, entry) } }) - ei.locks[key] = entry + ei.locks[scope] = entry } -func (ei *EventIsolation) releaseLocked(key eventIsolationKey, entry *lockEntry) { +func (ei *EventIsolation) releaseLocked(scope isolation.Scope, entry *lockEntry) { entry.timer.Stop() - delete(ei.locks, key) + delete(ei.locks, scope) close(entry.done) }