Merge d34eb3884d6510fe2b259bb574efd06077f42e49 into c97d63754b281655f1c54510a3cf6039c4fb067b

* fix: performance issues with Kafka caused by encapsulating the MQ interface

* fix: admin token in standalone mode

* fix: full id version

* fix: resolve deadlock in cache eviction and improve GetBatch implementation

CHANGELOG/CHANGELOG-3.8.md

@@ -1,70 +1,8 @@
-## [v3.8.3-patch.6](https://github.com/openimsdk/open-im-server/releases/tag/v3.8.3-patch.6) 	(2025-07-23)
+## [v3.8.3-patch.12](https://github.com/openimsdk/open-im-server/releases/tag/v3.8.3-patch.12) 	(2025-10-24)
 
 ### Bug Fixes
-* fix: Add friend DB in notification sender [#3438](https://github.com/openimsdk/open-im-server/pull/3438)
-* fix: remove update version file workflows have new line in 3.8.3-patch branch. [#3452](https://github.com/openimsdk/open-im-server/pull/3452)
-* fix: s3 aws init [#3454](https://github.com/openimsdk/open-im-server/pull/3454)
-* fix: use safe submodule init in workflows in v3.8.3-patch. [#3469](https://github.com/openimsdk/open-im-server/pull/3469)
+* fix: full id version [#3589](https://github.com/openimsdk/open-im-server/pull/3589)
+* fix: incorrect redis fields. [#3602](https://github.com/openimsdk/open-im-server/pull/3602)
 
-**Full Changelog**: [v3.8.3-patch.5...v3.8.3-patch.6](https://github.com/openimsdk/open-im-server/compare/v3.8.3-patch.5...v3.8.3-patch.6)
+**Full Changelog**: [v3.8.3-patch.11...v3.8.3-patch.12](https://github.com/openimsdk/open-im-server/compare/v3.8.3-patch.11...v3.8.3-patch.12)
 
-## [v3.8.3-patch.5](https://github.com/openimsdk/open-im-server/releases/tag/v3.8.3-patch.5) 	(2025-06-10)
-
-### New Features
-* feat: optimize friend and group applications [#3396](https://github.com/openimsdk/open-im-server/pull/3396)
-
-### Bug Fixes
-* fix: solve unocrrect invite notification [Created [#3219](https://github.com/openimsdk/open-im-server/pull/3219)
-
-### Builds
-* build: update gomake version in dockerfile.[Patch branch] [#3416](https://github.com/openimsdk/open-im-server/pull/3416)
-
-**Full Changelog**: [v3.8.3...v3.8.3-patch.5](https://github.com/openimsdk/open-im-server/compare/v3.8.3...v3.8.3-patch.5)
-
-## [v3.8.3-patch.4](https://github.com/openimsdk/open-im-server/releases/tag/v3.8.3-patch.4) 	(2025-03-13)
-
-### Bug Fixes
-* fix: solve unocrrect invite notificationfrom #3213
-
-**Full Changelog**: [v3.8.3-patch.3...v3.8.3-patch.4](https://github.com/openimsdk/open-im-server/compare/v3.8.3-patch.3...v3.8.3-patch.4)
-
-## [v3.8.3-patch.3](https://github.com/openimsdk/open-im-server/releases/tag/v3.8.3-patch.3) 	(2025-03-07)
-
-### New Features
-* feat: optimizing BatchGetIncrementalGroupMember #3180
-
-### Bug Fixes
-* fix: solve uncorrect notification when set group info #3172
-* fix: the sorting is wrong after canceling the administrator in group settings #3185
-* fix: solve uncorrect GroupMember enter group notification type. #3188
-
-### Refactors
-* refactor: change sendNotification to sendMessage to avoid ambiguity regarding message sending behavior. #3173
-
-**Full Changelog**: [v3.8.3-patch.2...v3.8.3-patch.3](https://github.com/openimsdk/open-im-server/compare/v3.8.3-patch.2...v3.8.3-patch.3)
-
-## [v3.8.3-patch.2](https://github.com/openimsdk/open-im-server/releases/tag/v3.8.3-patch.2) 	(2025-02-28)
-
-### Bug Fixes
-* fix: Offline push does not have a badge && Android offline push (#3146) [#3174](https://github.com/openimsdk/open-im-server/pull/3174)
-
-**Full Changelog**: [v3.8.3-patch.1...v3.8.3-patch.2](https://github.com/openimsdk/open-im-server/compare/v3.8.3-patch.1...v3.8.3-patch.2)
-
-## [v3.8.3-patch.1](https://github.com/openimsdk/open-im-server/releases/tag/v3.8.3-patch.1) 	(2025-02-25)
-
-### New Features
-* feat: add backup volume && optimize log print [Created [#3121](https://github.com/openimsdk/open-im-server/pull/3121)
-
-### Bug Fixes
-* fix: seq conversion failed without exiting [Created [#3120](https://github.com/openimsdk/open-im-server/pull/3120)
-* fix: check error in BatchSetTokenMapByUidPid [Created [#3123](https://github.com/openimsdk/open-im-server/pull/3123)
-* fix: DeleteDoc crash [Created [#3124](https://github.com/openimsdk/open-im-server/pull/3124)
-* fix: the abnormal message has no sending time, causing the SDK to be abnormal [Created [#3126](https://github.com/openimsdk/open-im-server/pull/3126)
-* fix: crash caused [#3127](https://github.com/openimsdk/open-im-server/pull/3127)
-* fix: the user sets the conversation timer cleanup timestamp unit incorrectly [Created [#3128](https://github.com/openimsdk/open-im-server/pull/3128)
-* fix: seq conversion not reading env in docker environment [Created [#3131](https://github.com/openimsdk/open-im-server/pull/3131)
-
-### Builds
-* build: improve workflows contents. [Created [#3125](https://github.com/openimsdk/open-im-server/pull/3125)
-
-**Full Changelog**: [v3.8.3-e-v1.1.5...v3.8.3-patch.1-e-v1.1.5](https://github.com/openimsdk/open-im-server-enterprise/compare/v3.8.3-e-v1.1.5...v3.8.3-patch.1-e-v1.1.5)

internal/msggateway/hub_server.go

@@ -152,19 +152,16 @@ func (s *Server) pushToUser(ctx context.Context, userID string, msgData *sdkws.M
 		userPlatform := &msggateway.SingleMsgToUserPlatform{
 			RecvPlatFormID: int32(client.PlatformID),
 		}
-		if !client.IsBackground ||
-			(client.IsBackground && client.PlatformID != constant.IOSPlatformID) {
-			err := client.PushMessage(ctx, msgData)
-			if err != nil {
-				log.ZWarn(ctx, "online push msg failed", err, "userID", userID, "platformID", client.PlatformID)
-				userPlatform.ResultCode = int64(servererrs.ErrPushMsgErr.Code())
-			} else {
-				if _, ok := s.pushTerminal[client.PlatformID]; ok {
-					result.OnlinePush = true
-				}
-			}
-		} else {
+		if client.IsBackground && client.PlatformID == constant.IOSPlatformID {
 			userPlatform.ResultCode = int64(servererrs.ErrIOSBackgroundPushErr.Code())
+			result.Resp = append(result.Resp, userPlatform)
+			continue
+		}
+		if err := client.PushMessage(ctx, msgData); err != nil {
+			log.ZWarn(ctx, "online push msg failed", err, "userID", userID, "platformID", client.PlatformID)
+			userPlatform.ResultCode = int64(servererrs.ErrPushMsgErr.Code())
+		} else if _, ok := s.pushTerminal[client.PlatformID]; ok {
+			result.OnlinePush = true
 		}
 		result.Resp = append(result.Resp, userPlatform)
 	}

pkg/localcache/cache.go

@@ -47,15 +47,15 @@ func New[V any](opts ...Option) Cache[V] {
 	if opt.localSlotNum > 0 && opt.localSlotSize > 0 {
 		createSimpleLRU := func() lru.LRU[string, V] {
 			if opt.expirationEvict {
-				return lru.NewExpirationLRU(opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict)
+				return lru.NewExpirationLRU[string, V](opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict)
 			} else {
-				return lru.NewLazyLRU(opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict)
+				return lru.NewLazyLRU[string, V](opt.localSlotSize, opt.localSuccessTTL, opt.localFailedTTL, opt.target, c.onEvict)
 			}
 		}
 		if opt.localSlotNum == 1 {
 			c.local = createSimpleLRU()
 		} else {
-			c.local = lru.NewSlotLRU(opt.localSlotNum, LRUStringHash, createSimpleLRU)
+			c.local = lru.NewSlotLRU[string, V](opt.localSlotNum, LRUStringHash, createSimpleLRU)
 		}
 		if opt.linkSlotNum > 0 {
 			c.link = link.New(opt.linkSlotNum)
@@ -71,14 +71,19 @@ type cache[V any] struct {
 }
 
 func (c *cache[V]) onEvict(key string, value V) {
-	_ = value
-
 	if c.link != nil {
-		lks := c.link.Del(key)
-		for k := range lks {
-			if key != k { // prevent deadlock
-				c.local.Del(k)
-			}
+		// Do not delete other keys while the underlying LRU still holds its lock;
+		// defer linked deletions to avoid re-entering the same slot and deadlocking.
+		if lks := c.link.Del(key); len(lks) > 0 {
+			go c.delLinked(key, lks)
+		}
+	}
+}
+
+func (c *cache[V]) delLinked(src string, keys map[string]struct{}) {
+	for k := range keys {
+		if src != k {
+			c.local.Del(k)
 		}
 	}
 }
@@ -105,7 +110,7 @@ func (c *cache[V]) Get(ctx context.Context, key string, fetch func(ctx context.C
 func (c *cache[V]) GetLink(ctx context.Context, key string, fetch func(ctx context.Context) (V, error), link ...string) (V, error) {
 	if c.local != nil {
 		return c.local.Get(key, func() (V, error) {
-			if len(link) > 0 {
+			if len(link) > 0 && c.link != nil {
 				c.link.Link(key, link...)
 			}
 			return fetch(ctx)

pkg/localcache/cache_test.go

@@ -22,6 +22,8 @@ import (
 	"sync/atomic"
 	"testing"
 	"time"
+
+	"github.com/openimsdk/open-im-server/v3/pkg/localcache/lru"
 )
 
 func TestName(t *testing.T) {
@@ -91,3 +93,68 @@ func TestName(t *testing.T) {
 	t.Log("del", del.Load())
 	// 137.35s
 }
+
+// Test deadlock scenario when eviction callback deletes a linked key that hashes to the same slot.
+func TestCacheEvictDeadlock(t *testing.T) {
+	ctx := context.Background()
+	c := New[string](WithLocalSlotNum(1), WithLocalSlotSize(1), WithLazy())
+
+	if _, err := c.GetLink(ctx, "k1", func(ctx context.Context) (string, error) {
+		return "v1", nil
+	}, "k2"); err != nil {
+		t.Fatalf("seed cache failed: %v", err)
+	}
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		_, _ = c.GetLink(ctx, "k2", func(ctx context.Context) (string, error) {
+			return "v2", nil
+		}, "k1")
+	}()
+
+	select {
+	case <-done:
+		// expected to finish quickly; current implementation deadlocks here.
+	case <-time.After(time.Second):
+		t.Fatal("GetLink deadlocked during eviction of linked key")
+	}
+}
+
+func TestExpirationLRUGetBatch(t *testing.T) {
+	l := lru.NewExpirationLRU[string, string](2, time.Minute, time.Second*5, EmptyTarget{}, nil)
+
+	keys := []string{"a", "b"}
+	values, err := l.GetBatch(keys, func(keys []string) (map[string]string, error) {
+		res := make(map[string]string)
+		for _, k := range keys {
+			res[k] = k + "_v"
+		}
+		return res, nil
+	})
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if len(values) != len(keys) {
+		t.Fatalf("expected %d values, got %d", len(keys), len(values))
+	}
+	for _, k := range keys {
+		if v, ok := values[k]; !ok || v != k+"_v" {
+			t.Fatalf("unexpected value for %s: %q, ok=%v", k, v, ok)
+		}
+	}
+
+	// second batch should hit cache
+	values, err = l.GetBatch(keys, func(keys []string) (map[string]string, error) {
+		t.Fatalf("should not fetch on cache hit")
+		return nil, nil
+	})
+	if err != nil {
+		t.Fatalf("unexpected error on cache hit: %v", err)
+	}
+	for _, k := range keys {
+		if v, ok := values[k]; !ok || v != k+"_v" {
+			t.Fatalf("unexpected cached value for %s: %q, ok=%v", k, v, ok)
+		}
+	}
+}

pkg/localcache/init.go

@@ -33,10 +33,6 @@ func InitLocalCache(localCache *config.LocalCache) {
 			Local config.CacheConfig
 			Keys  []string
 		}{
-			{
-				Local: localCache.Auth,
-				Keys:  []string{cachekey.UidPidToken},
-			},
 			{
 				Local: localCache.User,
 				Keys:  []string{cachekey.UserInfoKey, cachekey.UserGlobalRecvMsgOptKey},

pkg/localcache/lru/lru_expiration.go

@@ -52,8 +52,53 @@ type ExpirationLRU[K comparable, V any] struct {
 }
 
 func (x *ExpirationLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) (map[K]V, error) {
-	//TODO implement me
-	panic("implement me")
+	var (
+		err     error
+		results = make(map[K]V)
+		misses  = make([]K, 0, len(keys))
+	)
+
+	for _, key := range keys {
+		x.lock.Lock()
+		v, ok := x.core.Get(key)
+		x.lock.Unlock()
+		if ok {
+			x.target.IncrGetHit()
+			v.lock.RLock()
+			results[key] = v.value
+			if v.err != nil && err == nil {
+				err = v.err
+			}
+			v.lock.RUnlock()
+			continue
+		}
+		misses = append(misses, key)
+	}
+
+	if len(misses) == 0 {
+		return results, err
+	}
+
+	fetchValues, fetchErr := fetch(misses)
+	if fetchErr != nil && err == nil {
+		err = fetchErr
+	}
+
+	for key, val := range fetchValues {
+		results[key] = val
+		if fetchErr != nil {
+			x.target.IncrGetFailed()
+			continue
+		}
+		x.target.IncrGetSuccess()
+		item := &expirationLruItem[V]{value: val}
+		x.lock.Lock()
+		x.core.Add(key, item)
+		x.lock.Unlock()
+	}
+
+	// any keys not returned from fetch remain absent (no cache write)
+	return results, err
 }
 
 func (x *ExpirationLRU[K, V]) Get(key K, fetch func() (V, error)) (V, error) {

pkg/localcache/lru/lru_slot.go

@@ -35,7 +35,7 @@ type slotLRU[K comparable, V any] struct {
 func (x *slotLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)) (map[K]V, error) {
 	var (
 		slotKeys = make(map[uint64][]K)
-		kVs       = make(map[K]V)
+		vs       = make(map[K]V)
 	)
 
 	for _, k := range keys {
@@ -49,10 +49,10 @@ func (x *slotLRU[K, V]) GetBatch(keys []K, fetch func(keys []K) (map[K]V, error)
 			return nil, err
 		}
 		for key, value := range batches {
-			kVs[key] = value
+			vs[key] = value
 		}
 	}
-	return kVs, nil
+	return vs, nil
 }
 
 func (x *slotLRU[K, V]) getIndex(k K) uint64 {