diff --git a/datastore/cache/cache.go b/datastore/cache/cache.go index 787ad2803..e83ddcebd 100644 --- a/datastore/cache/cache.go +++ b/datastore/cache/cache.go @@ -24,16 +24,21 @@ type Cache struct { data *pendingmap.PendingMap flow flow.Flow - onlyCollectionField bool - collectionField dskey.Key + fullMessagebus bool } // New creates an initialized cache instance. -func New(flow flow.Flow) *Cache { - return &Cache{ +func New(flow flow.Flow, options ...Options) *Cache { + c := Cache{ data: pendingmap.New(), flow: flow, } + + for _, option := range options { + option(&c) + } + + return &c } // Get returns the values for a list of keys. If one or more keys do not exist @@ -74,6 +79,14 @@ func (c *Cache) Get(ctx context.Context, keys ...dskey.Key) (map[dskey.Key][]byt return got, nil } +// Snapshot returns a shnapshot over all data in the cache as a Getter. +func (c *Cache) Snapshot(notFoundHandler flow.Getter) Snapshot { + return Snapshot{ + data: c.data.Snapshot(), + notFoundHandler: notFoundHandler, + } +} + // fetchMissing loads all keys, that are currently not in the cache. // // Possible Errors: context.Canceled or context.DeadlineExeeded. @@ -132,6 +145,11 @@ func (c *Cache) Update(ctx context.Context, updateFn func(map[dskey.Key][]byte, c.data.SetIfPendingOrExists(data) } + if c.fullMessagebus { + c.data.Set(data) + } else { + c.data.SetIfPendingOrExists(data) + } updateFn(data, err) }) } diff --git a/datastore/cache/options.go b/datastore/cache/options.go new file mode 100644 index 000000000..7a739d95d --- /dev/null +++ b/datastore/cache/options.go @@ -0,0 +1,11 @@ +package cache + +// Options is a type for possible options to initialize the cache with. +type Options func(*Cache) + +// WithFullMessagebus sets the cache to use all messages from the messagebus. +// +// The default is, to use only messages with keys, that are already in the cache. +func WithFullMessagebus(c *Cache) { + c.fullMessagebus = true +} diff --git a/datastore/cache/pendingmap/pendingmap.go b/datastore/cache/pendingmap/pendingmap.go index 8ea150a39..ade94e465 100644 --- a/datastore/cache/pendingmap/pendingmap.go +++ b/datastore/cache/pendingmap/pendingmap.go @@ -6,6 +6,7 @@ import ( "sync" "github.com/OpenSlides/openslides-go/datastore/dskey" + "github.com/benbjohnson/immutable" ) // ErrNotExist is returned from pendingmap.Get() when a key was not pending at @@ -40,14 +41,14 @@ var ErrNotExist = errors.New("key does not exist") // pending. SetEmptyIfPending() sets a value to its zero value if it is pending. type PendingMap struct { mu sync.RWMutex - data map[dskey.Key][]byte + data *immutable.Map[dskey.Key, []byte] pending map[dskey.Key]chan struct{} } // New initializes a pendingDict. func New() *PendingMap { return &PendingMap{ - data: make(map[dskey.Key][]byte), + data: immutable.NewMap[dskey.Key, []byte](nil), pending: make(map[dskey.Key]chan struct{}), } } @@ -75,7 +76,7 @@ func (pm *PendingMap) Get(ctx context.Context, keys ...dskey.Key) (map[dskey.Key out := make(map[dskey.Key][]byte, len(keys)) err := pm.reading(func() error { for _, k := range keys { - v, ok := pm.data[k] + v, ok := pm.data.Get(k) if !ok { return ErrNotExist } @@ -90,6 +91,14 @@ func (pm *PendingMap) Get(ctx context.Context, keys ...dskey.Key) (map[dskey.Key return out, nil } +// Snapshot returns a snapshot of the data inside the pending map. +func (pm *PendingMap) Snapshot() *immutable.Map[dskey.Key, []byte] { + pm.mu.RLock() + defer pm.mu.RUnlock() + + return pm.data +} + // waitForPending blocks until all the given keys are not pending anymore. // // Expects, that all keys are either pending or in the data. It is not allowed, @@ -140,7 +149,7 @@ func (pm *PendingMap) MarkPending(keys ...dskey.Key) []dskey.Key { var needMark []dskey.Key pm.reading(func() error { for _, key := range keys { - if _, inStore := pm.data[key]; inStore { + if _, inStore := pm.data.Get(key); inStore { continue } if _, isPending := pm.pending[key]; isPending { @@ -166,7 +175,7 @@ func (pm *PendingMap) MarkPending(keys ...dskey.Key) []dskey.Key { continue } - if _, inStore := pm.data[key]; inStore { + if _, inStore := pm.data.Get(key); inStore { // The other caller has already the data continue } @@ -185,7 +194,7 @@ func (pm *PendingMap) UnMarkPending(keys ...dskey.Key) { defer pm.mu.Unlock() for _, key := range keys { - if _, ok := pm.data[key]; ok { + if _, ok := pm.data.Get(key); ok { continue } pending := pm.pending[key] @@ -199,7 +208,25 @@ func (pm *PendingMap) UnMarkPending(keys ...dskey.Key) { } } -// SetIfPendingOrExists updates values, but only if the key already exists or is pending. +// Set updates values. +// +// Informs all listeners. +func (pm *PendingMap) Set(data map[dskey.Key][]byte) { + pm.mu.Lock() + defer pm.mu.Unlock() + + for key, value := range data { + pm.data = pm.data.Set(key, value) + + if pending, isPending := pm.pending[key]; isPending { + close(pending) + delete(pm.pending, key) + } + } +} + +// SetIfPendingOrExists updates values, but only if the key already exists or is +// pending. // // If the key is pending, it is unmarked and all listeners are informed. func (pm *PendingMap) SetIfPendingOrExists(data map[dskey.Key][]byte) { @@ -208,13 +235,13 @@ func (pm *PendingMap) SetIfPendingOrExists(data map[dskey.Key][]byte) { for key, value := range data { pending := pm.pending[key] - _, exists := pm.data[key] + _, exists := pm.data.Get(key) if pending == nil && !exists { continue } - pm.data[key] = value + pm.data = pm.data.Set(key, value) if pending != nil { close(pending) @@ -232,7 +259,7 @@ func (pm *PendingMap) SetIfPending(data map[dskey.Key][]byte) { for key, value := range data { if pending, isPending := pm.pending[key]; isPending { - pm.data[key] = value + pm.data = pm.data.Set(key, value) close(pending) delete(pm.pending, key) } @@ -244,7 +271,7 @@ func (pm *PendingMap) Reset() { pm.mu.Lock() defer pm.mu.Unlock() - pm.data = make(map[dskey.Key][]byte) + pm.data = immutable.NewMap[dskey.Key, []byte](nil) pm.pending = make(map[dskey.Key]chan struct{}) } @@ -253,7 +280,7 @@ func (pm *PendingMap) Len() int { pm.mu.RLock() defer pm.mu.RUnlock() - return len(pm.data) + return pm.data.Len() } func (pm *PendingMap) reading(cmd func() error) error { @@ -269,7 +296,9 @@ func (pm *PendingMap) Size() int { defer pm.mu.RUnlock() var size int - for _, v := range pm.data { + itr := pm.data.Iterator() + for !itr.Done() { + _, v, _ := itr.Next() size += len(v) } return size diff --git a/datastore/cache/snapshot.go b/datastore/cache/snapshot.go new file mode 100644 index 000000000..b5d95ffbf --- /dev/null +++ b/datastore/cache/snapshot.go @@ -0,0 +1,56 @@ +package cache + +import ( + "context" + "fmt" + "maps" + + "github.com/OpenSlides/openslides-go/datastore/dskey" + "github.com/OpenSlides/openslides-go/datastore/flow" + "github.com/benbjohnson/immutable" +) + +// Snapshot implements the flow.Getter interface over an immutable map. +type Snapshot struct { + data *immutable.Map[dskey.Key, []byte] + notFoundHandler flow.Getter +} + +// Get returns keys from the snapshot. +func (s Snapshot) Get(ctx context.Context, keys ...dskey.Key) (map[dskey.Key][]byte, error) { + out := make(map[dskey.Key][]byte, len(keys)) + var notFound []dskey.Key + for _, key := range keys { + value, ok := s.data.Get(key) + if !ok { + notFound = append(notFound, key) + continue + } + out[key] = value + } + + if len(notFound) > 0 { + if s.notFoundHandler == nil { + return nil, IncompleteSnapshotError{notFound} + } + found, err := s.notFoundHandler.Get(ctx, notFound...) + if err != nil { + return nil, fmt.Errorf("not found handler: %w", err) + } + + maps.Copy(out, found) + } + + return out, nil +} + +// IncompleteSnapshotError is returned when a snapshot is ask for data, that does +// not exist in the snapshot. +type IncompleteSnapshotError struct { + Keys []dskey.Key +} + +// Error returns a string representation of the error. +func (err IncompleteSnapshotError) Error() string { + return fmt.Sprintf("did not found %d keys in snapshot", len(err.Keys)) +} diff --git a/go.mod b/go.mod index 6a05697b7..21675adc3 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module github.com/OpenSlides/openslides-go go 1.26.0 require ( + github.com/benbjohnson/immutable v0.4.3 github.com/goccy/go-yaml v1.19.2 github.com/golang-jwt/jwt/v4 v4.5.2 github.com/gomodule/redigo v1.9.3 @@ -42,5 +43,6 @@ require ( go.opentelemetry.io/otel v1.43.0 // indirect go.opentelemetry.io/otel/metric v1.43.0 // indirect go.opentelemetry.io/otel/trace v1.43.0 // indirect + golang.org/x/exp v0.0.0-20260218203240-3dfff04db8fa // indirect golang.org/x/text v0.37.0 // indirect ) diff --git a/go.sum b/go.sum index a3c5dfbd6..4425366af 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,7 @@ github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= +github.com/benbjohnson/immutable v0.4.3 h1:GYHcksoJ9K6HyAUpGxwZURrbTkXA0Dh4otXGqbhdrjA= +github.com/benbjohnson/immutable v0.4.3/go.mod h1:qJIKKSmdqz1tVzNtst1DZzvaqOU1onk1rc03IeM3Owk= github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= @@ -85,6 +87,8 @@ go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfC go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A= go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= +golang.org/x/exp v0.0.0-20260218203240-3dfff04db8fa h1:Zt3DZoOFFYkKhDT3v7Lm9FDMEV06GpzjG2jrqW+QTE0= +golang.org/x/exp v0.0.0-20260218203240-3dfff04db8fa/go.mod h1:K79w1Vqn7PoiZn+TkNpx3BUWUQksGO3JcVX6qIjytmA= golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY=