Fixed hot cache update on calls to Set()

Author: thrawn01

.gitignore

@@ -1,4 +1,6 @@
 *~
 .idea/
 .DS_Store
-vendor
+vendor
+.aider*
+.env

Makefile

@@ -6,7 +6,8 @@ $(GOLANGCI_LINT): ## Download Go linter
 
 .PHONY: ci
 ci: tidy lint test
-	go mod tidy && git diff --exit-code
+	@echo
+	@echo "\033[32mEVERYTHING PASSED!\033[0m"
 
 .PHONY: lint
 lint: $(GOLANGCI_LINT) ## Run Go linter

README.md

@@ -61,7 +61,8 @@ func ExampleUsage() {
     defer cancel()
 
     // ListenAndServe is a convenience function which Starts an instance of groupcache 
-    // with the provided transport and listens for groupcache HTTP requests on the address provided.
+    // with the provided transport and listens for groupcache HTTP requests on
+    // the address provided.
     d, err := groupcache.ListenAndServe(ctx, "192.168.1.1:8080", groupcache.Options{})
     if err != nil {
         log.Fatal("while starting server on 192.168.1.1:8080")

group.go

@@ -32,6 +32,8 @@ import (
 
 // Group is the user facing interface for a group
 type Group interface {
+	// TODO: deprecate the hotCache boolean in Set(). It is not needed
+
 	Set(context.Context, string, []byte, time.Time, bool) error
 	Get(context.Context, string, transport.Sink) error
 	Remove(context.Context, string) error
@@ -60,10 +62,10 @@ func (f GetterFunc) Get(ctx context.Context, key string, dest transport.Sink) er
 // A Group is a cache namespace and associated data loaded spread over
 // a group of 1 or more machines.
 type group struct {
-	name       string
-	getter     Getter
-	instance   *Instance
-	cacheBytes int64 // limit for sum of mainCache and hotCache size
+	name          string
+	getter        Getter
+	instance      *Instance
+	maxCacheBytes int64 // max size of both mainCache and hotCache
 
 	// mainCache is a cache of the keys for which this process
 	// (amongst its peers) is authoritative. That is, this cache
@@ -135,27 +137,56 @@ func (g *group) Get(ctx context.Context, key string, dest transport.Sink) error
 	return transport.SetSinkView(dest, value)
 }
 
-func (g *group) Set(ctx context.Context, key string, value []byte, expire time.Time, hotCache bool) error {
+func (g *group) Set(ctx context.Context, key string, value []byte, expire time.Time, _ bool) error {
 	if key == "" {
 		return errors.New("empty Set() key not allowed")
 	}
 
+	if g.maxCacheBytes <= 0 {
+		return nil
+	}
+
 	_, err := g.setGroup.Do(key, func() (interface{}, error) {
 		// If remote peer owns this key
 		owner, isRemote := g.instance.PickPeer(key)
 		if isRemote {
-			if err := g.setFromPeer(ctx, owner, key, value, expire); err != nil {
+			// Set the key/value on the remote peer
+			if err := g.setPeer(ctx, owner, key, value, expire); err != nil {
 				return nil, err
 			}
-			// TODO(thrawn01): Not sure if this is useful outside of tests...
-			//  maybe we should ALWAYS update the local cache?
-			if hotCache {
-				g.localSet(key, value, expire, g.hotCache)
+		}
+		// Update the local caches
+		bv := transport.ByteViewWithExpire(value, expire)
+		g.loadGroup.Lock(func() {
+			g.mainCache.Add(key, bv)
+			g.hotCache.Remove(key)
+		})
+
+		// Update all peers in the cluster
+		var wg sync.WaitGroup
+		for _, p := range g.instance.getAllPeers() {
+			if p.PeerInfo().IsSelf {
+				continue // Skip self
+			}
+
+			// Do not update the owner again, we already updated them
+			if p.HashKey() == owner.HashKey() {
+				continue
 			}
-			return nil, nil
+
+			wg.Add(1)
+			go func(p peer.Client) {
+				if err := g.setPeer(ctx, p, key, value, expire); err != nil {
+					g.instance.opts.Logger.Error("Failed to update peer",
+						"peer", p.PeerInfo().Address,
+						"key", key,
+						"err", err)
+				}
+				wg.Done()
+			}(p)
 		}
-		// We own this key
-		g.localSet(key, value, expire, g.mainCache)
+		wg.Wait()
+
 		return nil, nil
 	})
 	return err
@@ -340,7 +371,7 @@ func (g *group) getFromPeer(ctx context.Context, peer peer.Client, key string) (
 	return value, nil
 }
 
-func (g *group) setFromPeer(ctx context.Context, peer peer.Client, k string, v []byte, e time.Time) error {
+func (g *group) setPeer(ctx context.Context, peer peer.Client, k string, v []byte, e time.Time) error {
 	var expire int64
 	if !e.IsZero() {
 		expire = e.UnixNano()
@@ -363,7 +394,7 @@ func (g *group) removeFromPeer(ctx context.Context, peer peer.Client, key string
 }
 
 func (g *group) lookupCache(key string) (value transport.ByteView, ok bool) {
-	if g.cacheBytes <= 0 {
+	if g.maxCacheBytes <= 0 {
 		return
 	}
 	value, ok = g.mainCache.Get(key)
@@ -374,26 +405,30 @@ func (g *group) lookupCache(key string) (value transport.ByteView, ok bool) {
 	return
 }
 
-func (g *group) LocalSet(key string, value []byte, expire time.Time) {
-	g.localSet(key, value, expire, g.mainCache)
-}
-
-func (g *group) localSet(key string, value []byte, expire time.Time, cache Cache) {
-	if g.cacheBytes <= 0 {
+// RemoteSet is called by the transport to set values in the local and hot caches when
+// a remote peer sends us a pb.SetRequest
+func (g *group) RemoteSet(key string, value []byte, expire time.Time) {
+	if g.maxCacheBytes <= 0 {
 		return
 	}
 
-	bv := transport.ByteViewWithExpire(value, expire)
-
-	// Ensure no requests are in flight
+	// Lock all load operations until this function returns
 	g.loadGroup.Lock(func() {
-		g.populateCache(key, bv, cache)
+		// This instance could take over ownership of this key at any moment after
+		// the set is made. In order to avoid accidental propagation of the previous
+		// value should this instance become owner of the key, we always set key in
+		// the main cache.
+		bv := transport.ByteViewWithExpire(value, expire)
+		g.mainCache.Add(key, bv)
+
+		// It's possible the value could be in the hot cache.
+		g.hotCache.Remove(key)
 	})
 }
 
 func (g *group) LocalRemove(key string) {
 	// Clear key from our local cache
-	if g.cacheBytes <= 0 {
+	if g.maxCacheBytes <= 0 {
 		return
 	}
 
@@ -405,7 +440,7 @@ func (g *group) LocalRemove(key string) {
 }
 
 func (g *group) populateCache(key string, value transport.ByteView, cache Cache) {
-	if g.cacheBytes <= 0 {
+	if g.maxCacheBytes <= 0 {
 		return
 	}
 	cache.Add(key, value)
@@ -440,7 +475,7 @@ func (g *group) CacheStats(which CacheType) CacheStats {
 // ResetCacheSize changes the maxBytes allowed and resets both the main and hot caches.
 // It is mostly intended for testing and is not thread safe.
 func (g *group) ResetCacheSize(maxBytes int64) error {
-	g.cacheBytes = maxBytes
+	g.maxCacheBytes = maxBytes
 	var (
 		hotCache  int64
 		mainCache int64

instance.go

@@ -169,13 +169,13 @@ func (i *Instance) NewGroup(name string, cacheBytes int64, getter Getter) (Group
 		return nil, fmt.Errorf("duplicate registration of group '%s'", name)
 	}
 	g := &group{
-		instance:    i,
-		name:        name,
-		getter:      getter,
-		cacheBytes:  cacheBytes,
-		loadGroup:   &singleflight.Group{},
-		setGroup:    &singleflight.Group{},
-		removeGroup: &singleflight.Group{},
+		instance:      i,
+		name:          name,
+		getter:        getter,
+		maxCacheBytes: cacheBytes,
+		loadGroup:     &singleflight.Group{},
+		setGroup:      &singleflight.Group{},
+		removeGroup:   &singleflight.Group{},
 	}
 	if err := g.ResetCacheSize(cacheBytes); err != nil {
 		return nil, err

instance_test.go

@@ -33,6 +33,7 @@ import (
 	"github.com/groupcache/groupcache-go/v3/cluster"
 	"github.com/groupcache/groupcache-go/v3/transport"
 	"github.com/groupcache/groupcache-go/v3/transport/pb/testpb"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
 )
@@ -451,3 +452,46 @@ func TestNoDeDup(t *testing.T) {
 		t.Errorf("cache has %d bytes, want %d", used, wantBytes)
 	}
 }
+
+func TestSetValueOnAllPeers(t *testing.T) {
+	ctx := context.Background()
+	err := cluster.Start(ctx, 3, groupcache.Options{
+		Logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
+	})
+	require.NoError(t, err)
+	defer func() { _ = cluster.Shutdown(context.Background()) }()
+
+	// Create a group for each instance in the cluster
+	var groups []groupcache.Group
+	for _, d := range cluster.ListDaemons() {
+		g, err := d.GetInstance().NewGroup("group", 1<<20, groupcache.GetterFunc(func(ctx context.Context, key string, dest transport.Sink) error {
+			return dest.SetString("original-value", time.Time{})
+		}))
+		require.NoError(t, err)
+		groups = append(groups, g)
+	}
+
+	// Set the value on the first group
+	err = groups[0].Set(ctx, "key", []byte("value"), time.Time{}, false)
+	require.NoError(t, err)
+
+	// Verify the value exists on all peers
+	for i, g := range groups {
+		var result string
+		err := g.Get(ctx, "key", transport.StringSink(&result))
+		require.NoError(t, err, "Failed to get value from peer %d", i)
+		assert.Equal(t, "value", result, "Unexpected value from peer %d", i)
+	}
+
+	// Update the value on the second group
+	err = groups[1].Set(ctx, "key", []byte("foo"), time.Time{}, false)
+	require.NoError(t, err)
+
+	// Verify the value was updated
+	for i, g := range groups {
+		var result string
+		err := g.Get(ctx, "key", transport.StringSink(&result))
+		require.NoError(t, err, "Failed to get value from peer %d", i)
+		assert.Equal(t, "foo", result, "Unexpected value from peer %d", i)
+	}
+}

transport/http_transport.go

@@ -263,7 +263,7 @@ func (t *HttpTransport) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	type transportMethods interface {
 		Get(ctx context.Context, key string, dest Sink) error
-		LocalSet(string, []byte, time.Time)
+		RemoteSet(string, []byte, time.Time)
 		LocalRemove(string)
 	}
 
@@ -309,8 +309,7 @@ func (t *HttpTransport) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		if out.Expire != nil && *out.Expire != 0 {
 			expire = time.Unix(*out.Expire/int64(time.Second), *out.Expire%int64(time.Second))
 		}
-
-		group.LocalSet(*out.Key, out.Value, expire)
+		group.RemoteSet(*out.Key, out.Value, expire)
 		return
 	}