fix:duplicate cache keys caused by concurrent write and cache_value field type problem

.github/workflows/build-push.yml

@@ -5,9 +5,13 @@ on:
     branches:
       - "main"
       - "deploy/dev"
+      - "cache-test"
+      - "dify-for-mysql"
   pull_request:
     branches:
       - "main"
+      - "cache-test"
+      - "dify-for-mysql"
   release:
     types: [published]
 
@@ -16,9 +20,9 @@ concurrency:
   cancel-in-progress: true
 
 env:
-  DOCKERHUB_USER: ${{ secrets.DOCKERHUB_USER }}
-  DOCKERHUB_TOKEN: ${{ secrets.DOCKERHUB_TOKEN }}
-  DIFY_DAEMON_IMAGE_NAME: ${{ vars.DIFY_DAEMON_IMAGE_NAME || 'langgenius/dify-plugin-daemon' }}
+  DOCKERHUB_USER: ${{ secrets.DOCKER_USERNAME }}
+  DOCKERHUB_TOKEN: ${{ secrets.DOCKER_PASSWORD }}
+  DIFY_DAEMON_IMAGE_NAME: ${{ vars.DIFY_DAEMON_IMAGE_NAME || 'oceanbase/dify-plugin-daemon' }}
 
 jobs:
   matrix_prepare:
@@ -42,8 +46,8 @@ jobs:
 
   build:
     needs: matrix_prepare
-    runs-on: ${{ matrix.platform == 'linux/arm64' && 'arm64_runner' || 'ubuntu-latest' }}
-    if: github.repository == 'langgenius/dify-plugin-daemon'
+    #runs-on: ${{ matrix.platform == 'linux/arm64' && 'arm64_runner' || 'ubuntu-latest' }}
+    runs-on: ${{ matrix.platform == 'linux/arm64' && 'ubuntu-24.04-arm' || 'ubuntu-latest' }}
     strategy:
       matrix: ${{ fromJson(needs.matrix_prepare.outputs.matrix) }}
 
@@ -61,8 +65,8 @@ jobs:
         uses: docker/login-action@v3
         if: github.event_name != 'pull_request'
         with:
-          username: ${{ env.DOCKERHUB_USER }}
-          password: ${{ env.DOCKERHUB_TOKEN }}
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
 
       - name: Extract metadata (tags, labels) for Docker
         id: meta
@@ -105,14 +109,14 @@ jobs:
       - name: Login to Docker Hub
         uses: docker/login-action@v3
         with:
-          username: ${{ env.DOCKERHUB_USER }}
-          password: ${{ env.DOCKERHUB_TOKEN }}
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
 
       - name: Extract metadata (tags, labels) for Docker
         id: meta
         uses: docker/metadata-action@v5
         with:
-          images: ${{ vars.DIFY_DAEMON_IMAGE_NAME || 'langgenius/dify-plugin-daemon' }}
+          images: ${{ vars.DIFY_DAEMON_IMAGE_NAME || 'oceanbase/dify-plugin-daemon' }}
           tags: |
             type=raw,value=latest,enable=${{ startsWith(github.ref, 'refs/tags/') }}
             type=ref,event=branch

README.md

@@ -2,7 +2,7 @@
 
 ## Overview
 
-Dify Plugin Daemon is a service that manages the lifecycle of plugins. It's responsible for 3 types of runtimes:
+Dify Plugin Daemon is a service that manages the lifecycle of plugins. It's responsible for 3 types of runtimes,currently it is compatible for mysql and redis can be replaced with mysql cache:
 
 1. Local runtime: runs on the same machine as the Dify server.
 2. Debug runtime: listens to a port to wait for a debugging plugin to connect.
@@ -80,3 +80,4 @@ Refer to [Benchmark](https://langgenius.github.io/dify-plugin-daemon/benchmark-d
 ## LICENSE
 
 Dify Plugin Daemon is released under the [Apache-2.0 license](LICENSE).
+# Trigger build Tue Jul 15 02:34 PM CST 2025

docker/local.dockerfile

@@ -1,4 +1,6 @@
+# Use public image for GitHub Actions compatibility
 FROM golang:1.23-alpine AS builder
+# FROM reg.docker.alibaba-inc.com/dockerhub_common/golang:1.23-alpine AS builder
 
 ARG VERSION=unknown
 
@@ -9,7 +11,7 @@ COPY . /app
 WORKDIR /app
 
 # using goproxy if you have network issues
-# ENV GOPROXY=https://goproxy.cn,direct
+ENV GOPROXY=https://goproxy.cn,direct
 
 # build
 RUN go build \
@@ -22,15 +24,24 @@ RUN go build \
 COPY entrypoint.sh /app/entrypoint.sh
 RUN chmod +x /app/entrypoint.sh
 
-FROM ubuntu:24.04
+# Use public image for GitHub Actions compatibility
+FROM ubuntu:22.04
+# FROM reg.docker.alibaba-inc.com/ant-base/ubuntu:22.04
+USER root
+SHELL ["/bin/bash", "-c"]
 
 WORKDIR /app
 
 # check build args
 ARG PLATFORM=local
+ARG NEED_MIRROR=1
 
 # Install python3.12 if PLATFORM is local
-RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y curl python3.12 python3.12-venv python3.12-dev python3-pip ffmpeg build-essential \
+RUN apt-get upgrade && apt-get update && \
+    apt-get install -y software-properties-common && \
+    add-apt-repository ppa:deadsnakes/ppa -y && \
+    apt-get update && \
+    DEBIAN_FRONTEND=noninteractive apt-get install -y curl python3.12 python3.12-venv python3.12-dev python3-pip ffmpeg build-essential git unzip libssl-dev \
     && apt-get clean \
     && rm -rf /var/lib/apt/lists/* \
     && update-alternatives --install /usr/bin/python3 python3 /usr/bin/python3.12 1;
@@ -39,11 +50,23 @@ RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y curl pyt
 ENV TIKTOKEN_CACHE_DIR=/app/.tiktoken
 
 # Install dify_plugin to speedup the environment setup, test uv and preload tiktoken
-RUN mv /usr/lib/python3.12/EXTERNALLY-MANAGED /usr/lib/python3.12/EXTERNALLY-MANAGED.bk \
-    && python3 -m pip install uv \
+RUN [ -f /usr/lib/python3.12/EXTERNALLY-MANAGED ] && mv /usr/lib/python3.12/EXTERNALLY-MANAGED /usr/lib/python3.12/EXTERNALLY-MANAGED.bk || true \
+    && python3.12 -m ensurepip --upgrade \
+    && (if [ "$NEED_MIRROR" = "1" ]; then \
+        echo "Setting up Aliyun pip mirrors..." && \
+        python3.12 -m pip config set global.index-url https://mirrors.aliyun.com/pypi/simple && \
+        python3.12 -m pip config set global.trusted-host mirrors.aliyun.com && \
+        mkdir -p /etc/uv && \
+        echo "[[index]]" > /etc/uv/uv.toml && \
+        echo 'url = "https://mirrors.aliyun.com/pypi/simple"' >> /etc/uv/uv.toml && \
+        echo "default = true" >> /etc/uv/uv.toml && \
+        echo "Mirrors setup completed"; \
+    fi) \
+    && python3.12 -m pip install --upgrade pip setuptools wheel \
+    && python3.12 -m pip install uv \
     && uv pip install --system dify_plugin \
-    && python3 -c "from uv._find_uv import find_uv_bin;print(find_uv_bin());" \
-    && python3 -c "import tiktoken; encodings = ['o200k_base', 'cl100k_base', 'p50k_base', 'r50k_base', 'p50k_edit', 'gpt2']; [tiktoken.get_encoding(encoding).special_tokens_set for encoding in encodings]"
+    && python3.12 -c "from uv._find_uv import find_uv_bin;print(find_uv_bin());" \
+    && python3.12 -c "import tiktoken; encodings = ['o200k_base', 'cl100k_base', 'p50k_base', 'r50k_base', 'p50k_edit', 'gpt2']; [tiktoken.get_encoding(encoding).special_tokens_set for encoding in encodings]"
 
 ENV UV_PATH=/usr/local/bin/uv
 ENV PLATFORM=$PLATFORM

go.mod

@@ -7,6 +7,7 @@ toolchain go1.23.3
 require (
 	cloud.google.com/go/storage v1.51.0
 	github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.6.0
+	github.com/aliyun/aliyun-oss-go-sdk v3.0.2+incompatible
 	github.com/aws/aws-sdk-go-v2 v1.30.4
 	github.com/aws/aws-sdk-go-v2/config v1.27.31
 	github.com/aws/aws-sdk-go-v2/credentials v1.17.30
@@ -25,6 +26,7 @@ require (
 	github.com/stretchr/testify v1.10.0
 	github.com/tencentyun/cos-go-sdk-v5 v0.7.62
 	github.com/xeipuuv/gojsonschema v1.2.0
+	golang.org/x/oauth2 v0.28.0
 	google.golang.org/api v0.224.0
 	gorm.io/driver/mysql v1.5.7
 	gorm.io/gorm v1.25.11
@@ -44,7 +46,6 @@ require (
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.25.0 // indirect
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.51.0 // indirect
 	github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.51.0 // indirect
-	github.com/aliyun/aliyun-oss-go-sdk v3.0.2+incompatible // indirect
 	github.com/atotto/clipboard v0.1.4 // indirect
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.4 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.12 // indirect
@@ -127,7 +128,6 @@ require (
 	go.opentelemetry.io/otel/sdk v1.34.0 // indirect
 	go.opentelemetry.io/otel/sdk/metric v1.34.0 // indirect
 	go.opentelemetry.io/otel/trace v1.34.0 // indirect
-	golang.org/x/oauth2 v0.28.0 // indirect
 	golang.org/x/time v0.10.0 // indirect
 	google.golang.org/genproto v0.0.0-20250303144028-a0af3efb3deb // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20250303144028-a0af3efb3deb // indirect

internal/utils/cache/mysql/models.go

@@ -5,7 +5,7 @@ import "time"
 type CacheKV struct {
 	ID         int64     `json:"id" gorm:"column:id;primaryKey;type:bigint(20) auto_increment"`
 	CacheKey   string    `json:"cache_key" gorm:"column:cache_key;type:varchar(256);not null;unique"`
-	CacheValue []byte    `json:"cache_value" gorm:"column:cache_value;type:blob;not null"`
+	CacheValue []byte    `json:"cache_value" gorm:"column:cache_value;type:longblob;not null"`
 	ExpireTime time.Time `json:"expire_time" gorm:"index"`
 	CreatedAt  time.Time `json:"created_at"`
 	UpdatedAt  time.Time `json:"updated_at"`
@@ -15,7 +15,7 @@ type CacheMap struct {
 	ID         int64     `json:"id" gorm:"column:id;primaryKey;type:bigint(20) auto_increment"`
 	CacheKey   string    `json:"cache_key" gorm:"column:cache_key;type:varchar(256);not null;uniqueIndex:idx_cache_key_field"`
 	CacheField string    `json:"cache_field" gorm:"column:cache_field;type:varchar(256);not null;uniqueIndex:idx_cache_key_field"`
-	CacheValue string    `json:"cache_value" gorm:"column:cache_value;type:blob;not null"`
+	CacheValue string    `json:"cache_value" gorm:"column:cache_value;type:longblob;not null"`
 	CreatedAt  time.Time `json:"created_at"`
 	UpdatedAt  time.Time `json:"updated_at"`
 }

internal/utils/cache/mysql/mysql.go

@@ -67,15 +67,15 @@ func (c Client) Set(key string, value any, expire time.Duration) error {
 	val := toBytes(value)
 	expireTime := time.Now().Add(expire)
 
-	result := c.DB.Model(&CacheKV{}).
-		Where("cache_key = ?", key).
-		Assign(CacheKV{
-			CacheKey:   key,
-			CacheValue: val,
-			ExpireTime: expireTime,
-		}).
-		FirstOrCreate(&CacheKV{})
-	return result.Error
+	// 使用 INSERT ... ON DUPLICATE KEY UPDATE 来避免并发写入问题
+	sql := `INSERT INTO cache_kvs (cache_key, cache_value, expire_time, created_at, updated_at) 
+			VALUES (?, ?, ?, NOW(), NOW()) 
+			ON DUPLICATE KEY UPDATE 
+			cache_value = VALUES(cache_value), 
+			expire_time = VALUES(expire_time), 
+			updated_at = NOW()`
+
+	return c.DB.Exec(sql, key, val, expireTime).Error
 }
 
 func (c Client) GetBytes(key string) ([]byte, error) {
@@ -117,15 +117,14 @@ func (c Client) Count(key ...string) (int64, error) {
 }
 
 func (c Client) SetMapField(key string, field string, value string) error {
-	result := c.DB.Model(&CacheMap{}).
-		Where("cache_key = ? AND cache_field = ?", key, field).
-		Assign(CacheMap{
-			CacheKey:   key,
-			CacheField: field,
-			CacheValue: value,
-		}).
-		FirstOrCreate(&CacheMap{})
-	return result.Error
+	// 使用 INSERT ... ON DUPLICATE KEY UPDATE 来避免并发写入问题
+	sql := `INSERT INTO cache_maps (cache_key, cache_field, cache_value, created_at, updated_at) 
+			VALUES (?, ?, ?, NOW(), NOW()) 
+			ON DUPLICATE KEY UPDATE 
+			cache_value = VALUES(cache_value), 
+			updated_at = NOW()`
+
+	return c.DB.Exec(sql, key, field, value).Error
 }
 
 func (c Client) GetMapField(key string, field string) (string, error) {
@@ -198,24 +197,17 @@ func (c Client) SetNX(key string, value any, expire time.Duration) (bool, error)
 	val := toBytes(value)
 	expireTime := time.Now().Add(expire)
 
-	var existing CacheKV
-	result := c.DB.Where("cache_key = ?", key).First(&existing)
-	if result.Error == nil {
-		return false, nil
-	}
+	// 使用 INSERT IGNORE 来实现 SetNX，避免并发写入问题
+	sql := `INSERT IGNORE INTO cache_kvs (cache_key, cache_value, expire_time, created_at, updated_at) 
+			VALUES (?, ?, ?, NOW(), NOW())`
 
-	if result.Error.Error() != "record not found" {
+	result := c.DB.Exec(sql, key, val, expireTime)
+	if result.Error != nil {
 		return false, result.Error
 	}
 
-	newCache := CacheKV{
-		CacheKey:   key,
-		CacheValue: val,
-		ExpireTime: expireTime,
-	}
-
-	result = c.DB.Create(&newCache)
-	return result.Error == nil, result.Error
+	// 如果影响的行数为1，说明插入成功；如果为0，说明记录已存在
+	return result.RowsAffected == 1, nil
 }
 
 func (c Client) Expire(key string, expire time.Duration) (bool, error) {