Skip to content

fix: key session store by sync_id to prevent concurrent sync clobbering #976

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
c1-dev-bot wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix: key session store by sync_id to prevent concurrent sync clobbering #976

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 9 additions & 12 deletions internal/connector/connector.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,23 +68,20 @@ func (c *connectorClient) SetSessionStoreSetter(sessionStoreSetter sessions.SetS
c.sessionStoreSetter = sessionStoreSetter
}

func (c *connectorClient) SetSessionStore(ctx context.Context, store sessions.SessionStore) {
func (c *connectorClient) SetSessionStore(ctx context.Context, syncID string, store sessions.SessionStore) {
if c.sessionStoreSetter == nil {
// Demoted from Warn to Debug: this path is the normal case for
// any connector that didn't opt into session storage (i.e.,
// wrapper.run never set cw.SessionServer to non-nil, so
// SetSessionStoreSetter received nil at startup). The syncer
// still calls SetSessionStore unconditionally on every sync, so
// at Warn level this fires once per sync per non-session-store
// connector — pure noise that masked real warnings downstream.
// Debug keeps the "you forgot to wire this" signal available for
// developers running at debug level without polluting prod logs.
// See https://github.com/ConductorOne/baton-sdk/issues/907.
l := ctxzap.Extract(ctx)
l.Debug("connectorClient's session store is nil — connector did not opt into session storage")
return
}
c.sessionStoreSetter.SetSessionStore(ctx, store)
c.sessionStoreSetter.SetSessionStore(ctx, syncID, store)
}

func (c *connectorClient) RemoveSessionStore(ctx context.Context, syncID string) {
if c.sessionStoreSetter == nil {
return
}
c.sessionStoreSetter.RemoveSessionStore(ctx, syncID)
}

var ErrConnectorNotImplemented = errors.New("client does not implement connector connectorV2")
Expand Down
76 changes: 49 additions & 27 deletions pkg/session/session_server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
"fmt"
"log"
"net"
"sync"

v1 "github.com/conductorone/baton-sdk/pb/c1/connectorapi/baton/v1"
"github.com/conductorone/baton-sdk/pkg/types/sessions"
Expand All @@ -16,35 +17,50 @@

type GRPCSessionServer struct {
// v1.UnimplementedBatonSessionServiceServer
store sessions.SessionStore
mu sync.RWMutex
stores map[string]sessions.SessionStore
}

func NewGRPCSessionServer() *GRPCSessionServer {
return &GRPCSessionServer{}
return &GRPCSessionServer{
stores: make(map[string]sessions.SessionStore),
}
}

type SetSessionStore interface {
SetSessionStore(ctx context.Context, store sessions.SessionStore)
SetSessionStore(ctx context.Context, syncID string, store sessions.SessionStore)
RemoveSessionStore(ctx context.Context, syncID string)
}

func (s *GRPCSessionServer) SetSessionStore(ctx context.Context, store sessions.SessionStore) {
s.store = store
func (s *GRPCSessionServer) SetSessionStore(ctx context.Context, syncID string, store sessions.SessionStore) {
s.mu.Lock()
defer s.mu.Unlock()
s.stores[syncID] = store
}

func (s *GRPCSessionServer) Validate() error {
if s.store == nil {
return fmt.Errorf("session store is not set")
}
func (s *GRPCSessionServer) RemoveSessionStore(ctx context.Context, syncID string) {
s.mu.Lock()
defer s.mu.Unlock()
delete(s.stores, syncID)
}

return nil
func (s *GRPCSessionServer) getStore(syncID string) (sessions.SessionStore, error) {
s.mu.RLock()
defer s.mu.RUnlock()
store, ok := s.stores[syncID]
if !ok {
return nil, fmt.Errorf("session store not found for sync_id %q", syncID)
}
return store, nil
}

func (s *GRPCSessionServer) Get(ctx context.Context, req *v1.GetRequest) (*v1.GetResponse, error) {
if err := s.Validate(); err != nil {
store, err := s.getStore(req.GetSyncId())
if err != nil {
return nil, err
}

value, found, err := s.store.Get(ctx, req.GetKey(), sessions.WithSyncID(req.GetSyncId()), sessions.WithPrefix(req.GetPrefix()))
value, found, err := store.Get(ctx, req.GetKey(), sessions.WithSyncID(req.GetSyncId()), sessions.WithPrefix(req.GetPrefix()))
if err != nil {
return nil, fmt.Errorf("failed to get value from cache: %w", err)
}
Expand All @@ -56,11 +72,12 @@
}

func (s *GRPCSessionServer) GetMany(ctx context.Context, req *v1.GetManyRequest) (*v1.GetManyResponse, error) {
if err := s.Validate(); err != nil {
store, err := s.getStore(req.GetSyncId())
if err != nil {
return nil, err
}

values, unprocessedKeys, err := s.store.GetMany(
values, unprocessedKeys, err := store.GetMany(
ctx,
req.GetKeys(),
sessions.WithSyncID(req.GetSyncId()),
Expand All @@ -86,11 +103,12 @@
}

func (s *GRPCSessionServer) Set(ctx context.Context, req *v1.SetRequest) (*v1.SetResponse, error) {
if err := s.Validate(); err != nil {
store, err := s.getStore(req.GetSyncId())
if err != nil {
return nil, err
}

err := s.store.Set(ctx, req.GetKey(), req.GetValue(), sessions.WithSyncID(req.GetSyncId()), sessions.WithPrefix(req.GetPrefix()))
err = store.Set(ctx, req.GetKey(), req.GetValue(), sessions.WithSyncID(req.GetSyncId()), sessions.WithPrefix(req.GetPrefix()))
if err != nil {
return nil, fmt.Errorf("failed to set value in cache: %w", err)
}
Expand All @@ -99,11 +117,12 @@
}

func (s *GRPCSessionServer) SetMany(ctx context.Context, req *v1.SetManyRequest) (*v1.SetManyResponse, error) {
if err := s.Validate(); err != nil {
store, err := s.getStore(req.GetSyncId())
if err != nil {
return nil, err
}

err := s.store.SetMany(ctx, req.GetValues(), sessions.WithSyncID(req.GetSyncId()), sessions.WithPrefix(req.GetPrefix()))
err = store.SetMany(ctx, req.GetValues(), sessions.WithSyncID(req.GetSyncId()), sessions.WithPrefix(req.GetPrefix()))
if err != nil {
return nil, fmt.Errorf("failed to set many values in cache: %w", err)
}
Expand All @@ -112,11 +131,12 @@
}

func (s *GRPCSessionServer) Delete(ctx context.Context, req *v1.DeleteRequest) (*v1.DeleteResponse, error) {
if err := s.Validate(); err != nil {
store, err := s.getStore(req.GetSyncId())
if err != nil {
return nil, err
}

err := s.store.Delete(ctx, req.GetKey(), sessions.WithSyncID(req.GetSyncId()), sessions.WithPrefix(req.GetPrefix()))
err = store.Delete(ctx, req.GetKey(), sessions.WithSyncID(req.GetSyncId()), sessions.WithPrefix(req.GetPrefix()))
if err != nil {
return nil, fmt.Errorf("failed to delete value from cache: %w", err)
}
Expand All @@ -125,12 +145,13 @@
}

func (s *GRPCSessionServer) DeleteMany(ctx context.Context, req *v1.DeleteManyRequest) (*v1.DeleteManyResponse, error) {
if err := s.Validate(); err != nil {
store, err := s.getStore(req.GetSyncId())
if err != nil {
return nil, err
}

for _, key := range req.GetKeys() {
err := s.store.Delete(
err := store.Delete(
ctx,
key,
sessions.WithSyncID(req.GetSyncId()),
Expand All @@ -145,13 +166,13 @@
}

func (s *GRPCSessionServer) Clear(ctx context.Context, req *v1.ClearRequest) (*v1.ClearResponse, error) {
if s.store == nil {
// we sometimes clean up the session store after the connector is done
store, err := s.getStore(req.GetSyncId())
if err != nil {
ctxzap.Extract(ctx).Warn("session store is not set")
return &v1.ClearResponse{}, nil

Check failure on line 172 in pkg/session/session_server.go

View workflow job for this annotation

GitHub Actions / go-lint 

error is not nil (line 169) but it returns nil (nilerr)
}

err := s.store.Clear(ctx, sessions.WithSyncID(req.GetSyncId()), sessions.WithPrefix(req.GetPrefix()))
err = store.Clear(ctx, sessions.WithSyncID(req.GetSyncId()), sessions.WithPrefix(req.GetPrefix()))
if err != nil {
return nil, fmt.Errorf("failed to clear cache: %w", err)
}
Expand All @@ -160,11 +181,12 @@
}

func (s *GRPCSessionServer) GetAll(ctx context.Context, req *v1.GetAllRequest) (*v1.GetAllResponse, error) {
if err := s.Validate(); err != nil {
store, err := s.getStore(req.GetSyncId())
if err != nil {
return nil, err
}

values, nextPageToken, err := s.store.GetAll(
values, nextPageToken, err := store.GetAll(
ctx,
req.PageToken,
sessions.WithSyncID(req.GetSyncId()),
Expand Down
Loading
Loading