Feat: Add ZK config cache

common/constant/key.go

@@ -261,6 +261,7 @@ const (
 	ConfigSecretKey           = "config-center.secret"
 	ConfigBackupConfigKey     = "config-center.isBackupConfig"
 	ConfigBackupConfigPathKey = "config-center.backupConfigPath"
+	ConfigCacheTTLKey         = "config-center.cache-ttl"
 	ConfigRootPathParamKey    = "dubbo.config-center.root-path"
 )
 

config_center/zookeeper/config_cache.go

@@ -0,0 +1,237 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package zookeeper
+
+import (
+	"sync"
+	"time"
+)
+
+type configCacheEntry struct {
+	content   string
+	exists    bool
+	expiresAt time.Time
+}
+
+type configCache struct {
+	ttl time.Duration
+
+	stateLock  sync.RWMutex
+	entries    map[string]configCacheEntry
+	watches    map[string]bool
+	generation uint64
+
+	pathLocks sync.Map
+}
+
+func newConfigCache(ttl time.Duration) configCache {
+	return configCache{
+		ttl:     ttl,
+		entries: make(map[string]configCacheEntry),
+		watches: make(map[string]bool),
+	}
+}
+
+func (c *configCache) enabled() bool {
+	return c.ttl > 0
+}
+
+func (c *configCache) load(path string, loader func(bool) (configCacheEntry, bool, error)) (configCacheEntry, error) {
+	if !c.enabled() {
+		entry, _, err := loader(false)
+		return entry, err
+	}
+	if entry, ok := c.getFresh(path); ok {
+		return entry, nil
+	}
+
+	pathLock := c.pathLock(path)
+	pathLock.Lock()
+	defer pathLock.Unlock()
+
+	for {
+		if entry, ok := c.getFresh(path); ok {
+			return entry, nil
+		}
+
+		generation, watchActive := c.snapshot(path)
+		entry, nextWatchActive, err := loader(watchActive)
+		if err != nil {
+			if !c.storeWatchState(path, generation, nextWatchActive) {
+				continue
+			}
+			return configCacheEntry{}, err
+		}
+		if !c.storeLoad(path, generation, entry, nextWatchActive) {
+			continue
+		}
+		return entry, nil
+	}
+}
+
+func (c *configCache) store(path string, entry configCacheEntry) {
+	if !c.enabled() {
+		return
+	}
+	pathLock := c.pathLock(path)
+	pathLock.Lock()
+	defer pathLock.Unlock()
+
+	c.stateLock.Lock()
+	defer c.stateLock.Unlock()
+	c.storeEntryLocked(path, entry)
+}
+
+func (c *configCache) storeAtGeneration(path string, generation uint64, entry configCacheEntry) {
+	if !c.enabled() {
+		return
+	}
+	pathLock := c.pathLock(path)
+	pathLock.Lock()
+	defer pathLock.Unlock()
+
+	c.stateLock.Lock()
+	defer c.stateLock.Unlock()
+	if c.generation == generation {
+		c.storeEntryLocked(path, entry)
+	}
+}
+
+func (c *configCache) getFresh(path string) (configCacheEntry, bool) {
+	c.stateLock.RLock()
+	defer c.stateLock.RUnlock()
+
+	entry, ok := c.entries[path]
+	if !ok {
+		return configCacheEntry{}, false
+	}
+	return entry, entry.expiresAt.After(time.Now())
+}
+
+func (c *configCache) storeEntryLocked(path string, entry configCacheEntry) {
+	entry.expiresAt = time.Now().Add(c.ttl)
+	c.entries[path] = entry
+}
+
+func (c *configCache) snapshot(path string) (uint64, bool) {
+	c.stateLock.RLock()
+	defer c.stateLock.RUnlock()
+	return c.generation, c.watches[path]
+}
+
+func (c *configCache) isCurrentGeneration(generation uint64) bool {
+	c.stateLock.RLock()
+	defer c.stateLock.RUnlock()
+	return c.generation == generation
+}
+
+func (c *configCache) storeLoad(path string, generation uint64, entry configCacheEntry, watchActive bool) bool {
+	c.stateLock.Lock()
+	defer c.stateLock.Unlock()
+	if c.generation != generation {
+		return false
+	}
+
+	c.storeEntryLocked(path, entry)
+	c.setWatchActiveLocked(path, watchActive)
+	return true
+}
+
+func (c *configCache) storeWatchState(path string, generation uint64, watchActive bool) bool {
+	c.stateLock.Lock()
+	defer c.stateLock.Unlock()
+	if c.generation != generation {
+		return false
+	}
+	c.setWatchActiveLocked(path, watchActive)
+	return true
+}
+
+func (c *configCache) setWatchActive(path string, active bool) {
+	if !c.enabled() {
+		return
+	}
+	pathLock := c.pathLock(path)
+	pathLock.Lock()
+	defer pathLock.Unlock()
+
+	c.stateLock.Lock()
+	defer c.stateLock.Unlock()
+	c.setWatchActiveLocked(path, active)
+}
+
+func (c *configCache) setWatchActiveAtGeneration(path string, generation uint64, active bool) {
+	if !c.enabled() {
+		return
+	}
+	pathLock := c.pathLock(path)
+	pathLock.Lock()
+	defer pathLock.Unlock()
+	c.storeWatchState(path, generation, active)
+}
+
+func (c *configCache) setWatchActiveLocked(path string, active bool) {
+	if active {
+		c.watches[path] = true
+		return
+	}
+	delete(c.watches, path)
+}
+
+func (c *configCache) ensureWatch(path string, register func() error) error {
+	if !c.enabled() {
+		return register()
+	}
+	pathLock := c.pathLock(path)
+	pathLock.Lock()
+	defer pathLock.Unlock()
+
+	for {
+		generation, active := c.snapshot(path)
+		if active {
+			return nil
+		}
+		if err := register(); err != nil {
+			if !c.isCurrentGeneration(generation) {
+				continue
+			}
+			return err
+		}
+
+		c.stateLock.Lock()
+		if c.generation == generation {
+			c.watches[path] = true
+			c.stateLock.Unlock()
+			return nil
+		}
+		c.stateLock.Unlock()
+	}
+}
+
+func (c *configCache) reset() {
+	c.stateLock.Lock()
+	defer c.stateLock.Unlock()
+	c.generation++
+	c.entries = make(map[string]configCacheEntry)
+	c.watches = make(map[string]bool)
+}
+
+func (c *configCache) pathLock(path string) *sync.Mutex {
+	lock, _ := c.pathLocks.LoadOrStore(path, &sync.Mutex{})
+	return lock.(*sync.Mutex)
+}

config_center/zookeeper/config_cache_test.go

@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package zookeeper
+
+import (
+	"errors"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+import (
+	"github.com/stretchr/testify/require"
+)
+
+func TestConfigCacheLoadAndExpiry(t *testing.T) {
+	cache := newConfigCache(20 * time.Millisecond)
+	var loads atomic.Int32
+	var watchStates []bool
+	loader := func(watchActive bool) (configCacheEntry, bool, error) {
+		watchStates = append(watchStates, watchActive)
+		count := loads.Add(1)
+		return configCacheEntry{content: string(rune('0' + count)), exists: true}, true, nil
+	}
+
+	first, err := cache.load("/path", loader)
+	require.NoError(t, err)
+	second, err := cache.load("/path", loader)
+	require.NoError(t, err)
+	require.Equal(t, first.content, second.content)
+	require.Equal(t, int32(1), loads.Load())
+
+	require.Eventually(t, func() bool {
+		entry, loadErr := cache.load("/path", loader)
+		return loadErr == nil && entry.content == "2"
+	}, time.Second, 5*time.Millisecond)
+	require.Equal(t, int32(2), loads.Load())
+	require.Equal(t, []bool{false, true}, watchStates)
+
+	readErr := errors.New("read failed after watch registration")
+	_, err = cache.load("/error", func(bool) (configCacheEntry, bool, error) {
+		return configCacheEntry{}, true, readErr
+	})
+	require.ErrorIs(t, err, readErr)
+	_, watchActive := cache.snapshot("/error")
+	require.True(t, watchActive)
+}
+
+func TestConfigCacheWatchUpdateWinsOverLoad(t *testing.T) {
+	cache := newConfigCache(time.Minute)
+	loadStarted := make(chan struct{})
+	releaseLoad := make(chan struct{})
+	loadDone := make(chan struct{})
+	go func() {
+		defer close(loadDone)
+		_, _ = cache.load("/path", func(bool) (configCacheEntry, bool, error) {
+			close(loadStarted)
+			<-releaseLoad
+			return configCacheEntry{content: "old", exists: true}, true, nil
+		})
+	}()
+
+	<-loadStarted
+	updateDone := make(chan struct{})
+	go func() {
+		defer close(updateDone)
+		cache.store("/path", configCacheEntry{content: "new", exists: true})
+	}()
+	close(releaseLoad)
+	<-loadDone
+	<-updateDone
+
+	entry, ok := cache.getFresh("/path")
+	require.True(t, ok)
+	require.Equal(t, "new", entry.content)
+}
+
+func TestConfigCacheResetDiscardsInFlightLoad(t *testing.T) {
+	cache := newConfigCache(time.Minute)
+	cache.setWatchActive("/path", true)
+	loadStarted := make(chan struct{})
+	releaseLoad := make(chan struct{})
+	result := make(chan configCacheEntry, 1)
+	var loads atomic.Int32
+
+	go func() {
+		entry, _ := cache.load("/path", func(bool) (configCacheEntry, bool, error) {
+			if loads.Add(1) == 1 {
+				close(loadStarted)
+				<-releaseLoad
+				return configCacheEntry{content: "old", exists: true}, true, nil
+			}
+			return configCacheEntry{content: "new", exists: true}, true, nil
+		})
+		result <- entry
+	}()
+
+	<-loadStarted
+	cache.reset()
+	_, ok := cache.getFresh("/path")
+	require.False(t, ok)
+	_, watchActive := cache.snapshot("/path")
+	require.False(t, watchActive)
+	close(releaseLoad)
+
+	require.Equal(t, "new", (<-result).content)
+	require.Equal(t, int32(2), loads.Load())
+	entry, ok := cache.getFresh("/path")
+	require.True(t, ok)
+	require.Equal(t, "new", entry.content)
+}