3pp-mirror/pomerium
1
0
Fork 0
mirror of https://github.com/pomerium/pomerium.git synced 2025-07-25 20:49:30 +02:00
Code Issues Projects Releases Packages Wiki Activity

clean up context

Browse source
This commit is contained in:
Caleb Doxsey 2024-04-22 10:52:21 -06:00
parent 0e3e5ff494
commit a8efb5901e
2 changed files with 24 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

12
internal/identity/manager/manager.go
View file

@ -3,6 +3,7 @@ package manager
import ( import (
"context" "context"
"fmt"
"sync" "sync"
"time" "time"
@ -81,14 +82,14 @@ func (mgr *Manager) RunLeased(ctx context.Context) error {
}) })
eg, ctx := errgroup.WithContext(ctx) eg, ctx := errgroup.WithContext(ctx)
eg.Go(func() error { eg.Go(func() error {
sessionSyncer := newSessionSyncer(mgr) sessionSyncer := newSessionSyncer(ctx, mgr)
defer sessionSyncer.Close() defer sessionSyncer.Close()
return sessionSyncer.Run(ctx) return fmt.Errorf("session syncer error: %w", sessionSyncer.Run(ctx))
}) })
eg.Go(func() error { eg.Go(func() error {
userSyncer := newUserSyncer(mgr) userSyncer := newUserSyncer(ctx, mgr)
defer userSyncer.Close() defer userSyncer.Close()
return userSyncer.Run(ctx) return fmt.Errorf("user syncer error: %w", userSyncer.Run(ctx))
}) })
return eg.Wait() return eg.Wait()
} }
@ -296,7 +297,7 @@ func (mgr *Manager) updateUserInfo(ctx context.Context, userID string) {
continue continue
} }
err := authenticator.UpdateUserInfo(ctx, FromOAuthToken(s.GetOauthToken()), newMultiUnmarshaler(newUserUnmarshaler(u), newSessionUnmarshaler(s))) err := authenticator.UpdateUserInfo(ctx, FromOAuthToken(s.GetOauthToken()), newUserUnmarshaler(u))
metrics.RecordIdentityManagerUserRefresh(ctx, err) metrics.RecordIdentityManagerUserRefresh(ctx, err)
mgr.recordLastError(metrics_ids.IdentityManagerLastUserRefreshError, err) mgr.recordLastError(metrics_ids.IdentityManagerLastUserRefreshError, err)
if isTemporaryError(err) { if isTemporaryError(err) {
@ -314,7 +315,6 @@ func (mgr *Manager) updateUserInfo(ctx context.Context, userID string) {
continue continue
} }
mgr.updateSession(ctx, s)
mgr.updateUser(ctx, u) mgr.updateUser(ctx, u)
} }
} }

34
internal/identity/manager/sync.go
View file

@ -11,66 +11,68 @@ import (
) )
type sessionSyncerHandler struct { type sessionSyncerHandler struct {
ctx context.Context
mgr *Manager mgr *Manager
} }
func newSessionSyncer(mgr *Manager) *databroker.Syncer { func newSessionSyncer(ctx context.Context, mgr *Manager) *databroker.Syncer {
return databroker.NewSyncer("identity_manager/sessions", sessionSyncerHandler{mgr: mgr}, return databroker.NewSyncer("identity_manager/sessions", sessionSyncerHandler{ctx: ctx, mgr: mgr},
databroker.WithTypeURL(grpcutil.GetTypeURL(new(session.Session)))) databroker.WithTypeURL(grpcutil.GetTypeURL(new(session.Session))))
} }
func (h sessionSyncerHandler) ClearRecords(ctx context.Context) { func (h sessionSyncerHandler) ClearRecords(_ context.Context) {
h.mgr.onDeleteAllSessions(ctx) h.mgr.onDeleteAllSessions(h.ctx)
} }
func (h sessionSyncerHandler) GetDataBrokerServiceClient() databroker.DataBrokerServiceClient { func (h sessionSyncerHandler) GetDataBrokerServiceClient() databroker.DataBrokerServiceClient {
return h.mgr.cfg.Load().dataBrokerClient return h.mgr.cfg.Load().dataBrokerClient
} }
func (h sessionSyncerHandler) UpdateRecords(ctx context.Context, _ uint64, records []*databroker.Record) { func (h sessionSyncerHandler) UpdateRecords(_ context.Context, _ uint64, records []*databroker.Record) {
for _, record := range records { for _, record := range records {
if record.GetDeletedAt() != nil { if record.GetDeletedAt() != nil {
h.mgr.onDeleteSession(ctx, record.GetId()) h.mgr.onDeleteSession(h.ctx, record.GetId())
} else { } else {
var s session.Session var s session.Session
err := record.Data.UnmarshalTo(&s) err := record.Data.UnmarshalTo(&s)
if err != nil { if err != nil {
log.Ctx(ctx).Warn().Err(err).Msg("invalid data in session record, ignoring") log.Ctx(h.ctx).Warn().Err(err).Msg("invalid data in session record, ignoring")
} else { } else {
h.mgr.onUpdateSession(ctx, &s) h.mgr.onUpdateSession(h.ctx, &s)
} }
} }
} }
} }
type userSyncerHandler struct { type userSyncerHandler struct {
ctx context.Context
mgr *Manager mgr *Manager
} }
func newUserSyncer(mgr *Manager) *databroker.Syncer { func newUserSyncer(ctx context.Context, mgr *Manager) *databroker.Syncer {
return databroker.NewSyncer("identity_manager/users", userSyncerHandler{mgr: mgr}, return databroker.NewSyncer("identity_manager/users", userSyncerHandler{ctx: ctx, mgr: mgr},
databroker.WithTypeURL(grpcutil.GetTypeURL(new(user.User)))) databroker.WithTypeURL(grpcutil.GetTypeURL(new(user.User))))
} }
func (h userSyncerHandler) ClearRecords(ctx context.Context) { func (h userSyncerHandler) ClearRecords(_ context.Context) {
h.mgr.onDeleteAllUsers(ctx) h.mgr.onDeleteAllUsers(h.ctx)
} }
func (h userSyncerHandler) GetDataBrokerServiceClient() databroker.DataBrokerServiceClient { func (h userSyncerHandler) GetDataBrokerServiceClient() databroker.DataBrokerServiceClient {
return h.mgr.cfg.Load().dataBrokerClient return h.mgr.cfg.Load().dataBrokerClient
} }
func (h userSyncerHandler) UpdateRecords(ctx context.Context, _ uint64, records []*databroker.Record) { func (h userSyncerHandler) UpdateRecords(_ context.Context, _ uint64, records []*databroker.Record) {
for _, record := range records { for _, record := range records {
if record.GetDeletedAt() != nil { if record.GetDeletedAt() != nil {
h.mgr.onDeleteUser(ctx, record.GetId()) h.mgr.onDeleteUser(h.ctx, record.GetId())
} else { } else {
var u user.User var u user.User
err := record.Data.UnmarshalTo(&u) err := record.Data.UnmarshalTo(&u)
if err != nil { if err != nil {
log.Ctx(ctx).Warn().Err(err).Msg("invalid data in user record, ignoring") log.Ctx(h.ctx).Warn().Err(err).Msg("invalid data in user record, ignoring")
} else { } else {
h.mgr.onUpdateUser(ctx, &u) h.mgr.onUpdateUser(h.ctx, &u)
} }
} }
} }