Use immutable map for the cache

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)
 	})
 }

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
+}

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

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))
+}

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
 )

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=