Skip to content

Commit f272e6c

Browse files
fateleifatelei
authored
feat: install plugin should be idempotent (#576)
* feat: install plugin should be idempotent * chore: add new test
1 parent 35825a7 commit f272e6c

9 files changed

Lines changed: 148 additions & 21 deletions

File tree

‎internal/cluster/plugin.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ func (c *Cluster) RegisterPlugin(lifetime plugin_entities.PluginLifetime) error
4949
// do plugin state update immediately
5050
err = c.doPluginStateUpdate(l)
5151
if err != nil {
52-
return errors.Join(err, errors.New("failed to update plugin state"))
52+
return errors.Join(err, errors.New("failed to update plugin state"))
5353
}
5454

5555
if c.showLog {

‎internal/core/plugin_manager/packages.go‎

Lines changed: 25 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"errors"
55
"fmt"
66
"os"
7+
"strings"
78

89
"github.com/langgenius/dify-plugin-daemon/internal/db"
910
"github.com/langgenius/dify-plugin-daemon/internal/types/models"
@@ -56,17 +57,27 @@ func (p *PluginManager) SavePackage(
5657
return nil, err
5758
}
5859

59-
// create plugin if not exists
60+
// create plugin if not exists (idempotent under concurrency)
6061
if _, err := db.GetOne[models.PluginDeclaration](
6162
db.Equal("plugin_unique_identifier", uniqueIdentifier.String()),
6263
); err == db.ErrDatabaseNotFound {
63-
err = db.Create(&models.PluginDeclaration{
64+
createErr := db.Create(&models.PluginDeclaration{
6465
PluginUniqueIdentifier: uniqueIdentifier.String(),
6566
PluginID: uniqueIdentifier.PluginID(),
6667
Declaration: declaration,
6768
})
68-
if err != nil {
69-
return nil, err
69+
if createErr != nil {
70+
// ignore Postgres unique-violation (23505) errors triggered by concurrent inserts
71+
if isUniqueViolation(createErr) {
72+
return &declaration, nil
73+
}
74+
// fallback: if another goroutine has just inserted, read-after-write should succeed
75+
if _, again := db.GetOne[models.PluginDeclaration](
76+
db.Equal("plugin_unique_identifier", uniqueIdentifier.String()),
77+
); again == nil {
78+
return &declaration, nil
79+
}
80+
return nil, createErr
7081
}
7182
} else if err != nil {
7283
return nil, err
@@ -75,6 +86,16 @@ func (p *PluginManager) SavePackage(
7586
return &declaration, nil
7687
}
7788

89+
// isUniqueViolation returns true if err indicates a PostgreSQL unique constraint violation (SQLSTATE 23505).
90+
// Works across common drivers by matching canonical substrings to avoid hard dependency on driver types.
91+
func isUniqueViolation(err error) bool {
92+
if err == nil {
93+
return false
94+
}
95+
s := err.Error()
96+
return strings.Contains(s, "SQLSTATE 23505") || strings.Contains(s, "duplicate key value violates unique constraint")
97+
}
98+
7899
func (p *PluginManager) GetPackage(
79100
plugin_unique_identifier plugin_entities.PluginUniqueIdentifier,
80101
) ([]byte, error) {

‎internal/db/cache.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,4 +2,4 @@ package db
22

33
// Note: The GetCache, UpdateCache, and DeleteCache functions that were previously
44
// in this file are deprecated and not used in the codebase.
5-
// Direct cache operations should use the cache package (internal/utils/cache)
5+
// Direct cache operations should use the cache package (internal/utils/cache)

‎internal/service/unauthorized_langgenius_test.go‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -138,7 +138,7 @@ func TestIsUnauthorizedLanggenius(t *testing.T) {
138138
Author: tt.author,
139139
},
140140
}
141-
141+
142142
got := isUnauthorizedLanggenius(declaration, tt.verification)
143143
if got != tt.want {
144144
t.Errorf("isUnauthorizedLanggenius() = %v, want %v", got, tt.want)
@@ -163,10 +163,10 @@ func TestIsUnauthorizedLanggenius_EdgeCases(t *testing.T) {
163163
want: false, // spaces don't affect the comparison after lowercase
164164
},
165165
{
166-
name: "langgenius with spaces but no verification",
167-
author: " langgenius ",
166+
name: "langgenius with spaces but no verification",
167+
author: " langgenius ",
168168
verification: nil,
169-
want: false, // with spaces, not exact match after lowercase
169+
want: false, // with spaces, not exact match after lowercase
170170
},
171171
{
172172
name: "LaNgGeNiUs mixed case",
@@ -193,11 +193,11 @@ func TestIsUnauthorizedLanggenius_EdgeCases(t *testing.T) {
193193
Author: tt.author,
194194
},
195195
}
196-
196+
197197
got := isUnauthorizedLanggenius(declaration, tt.verification)
198198
if got != tt.want {
199199
t.Errorf("isUnauthorizedLanggenius() = %v, want %v for author=%q", got, tt.want, tt.author)
200200
}
201201
})
202202
}
203-
}
203+
}

‎internal/types/app/default_test.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -6,9 +6,9 @@ import (
66

77
func TestServerHostDefault(t *testing.T) {
88
tests := []struct {
9-
name string
10-
inputHost string
11-
expectedHost string
9+
name string
10+
inputHost string
11+
expectedHost string
1212
}{
1313
{
1414
name: "empty host should default to 0.0.0.0",

‎internal/types/models/curd/atomic.go‎

Lines changed: 19 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -61,10 +61,25 @@ func InstallPlugin(
6161

6262
err := db.Create(plugin, tx)
6363
if err != nil {
64-
return err
64+
// Handle potential duplicate creation due to race: refetch and update refers
65+
// to achieve idempotent behavior under concurrency.
66+
p2, gerr := db.GetOne[models.Plugin](
67+
db.WithTransactionContext(tx),
68+
db.Equal("plugin_unique_identifier", pluginUniqueIdentifier.String()),
69+
db.Equal("install_type", string(installType)),
70+
db.WLock(),
71+
)
72+
if gerr != nil {
73+
return err
74+
}
75+
p2.Refers++
76+
if uerr := db.Update(&p2, tx); uerr != nil {
77+
return uerr
78+
}
79+
pluginToBeReturns = &p2
80+
} else {
81+
pluginToBeReturns = plugin
6582
}
66-
67-
pluginToBeReturns = plugin
6883
} else if err != nil {
6984
return err
7085
} else {
@@ -604,7 +619,7 @@ func UpgradePlugin(
604619
if err != nil {
605620
return nil, err
606621
}
607-
pluginId := newPluginUniqueIdentifier.PluginID() // get the pluginId
622+
pluginId := newPluginUniqueIdentifier.PluginID() // get the pluginId
608623
pluginInstallationCacheKey := helper.PluginInstallationCacheKey(pluginId, tenantId) // make cache key
609624
if _, err = cache.AutoDelete[models.PluginInstallation](pluginInstallationCacheKey); err != nil {
610625
return nil, err

‎internal/types/models/curd/install_concurrency_test.go‎

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,90 @@
1+
package curd
2+
3+
import (
4+
"strings"
5+
"sync"
6+
"testing"
7+
8+
"github.com/google/uuid"
9+
"github.com/langgenius/dify-plugin-daemon/internal/db"
10+
"github.com/langgenius/dify-plugin-daemon/internal/types/app"
11+
"github.com/langgenius/dify-plugin-daemon/internal/types/models"
12+
"github.com/langgenius/dify-plugin-daemon/pkg/entities/plugin_entities"
13+
"github.com/stretchr/testify/require"
14+
)
15+
16+
// TestInstallPlugin_IdempotentUnderConcurrency ensures creating the same plugin/installation
17+
// concurrently is idempotent: only one plugin and one installation row are persisted.
18+
func TestInstallPlugin_IdempotentUnderConcurrency(t *testing.T) {
19+
cfg := &app.Config{
20+
DBType: app.DB_TYPE_POSTGRESQL,
21+
DBUsername: "postgres",
22+
DBPassword: "difyai123456",
23+
DBHost: "localhost",
24+
DBPort: 5432,
25+
DBDatabase: "dify_plugin_daemon",
26+
DBSslMode: "disable",
27+
}
28+
cfg.SetDefault()
29+
db.Init(cfg)
30+
t.Cleanup(db.Close)
31+
32+
tenantID := uuid.NewString()
33+
pluginName := "concurrency_demo_" + uuid.NewString()
34+
checksum := uuid.NewString()
35+
checksum = strings.ReplaceAll(checksum, "-", "")
36+
// 32 hex chars
37+
if len(checksum) > 32 {
38+
checksum = checksum[:32]
39+
}
40+
41+
identifier, err := plugin_entities.NewPluginUniqueIdentifier("tester/" + pluginName + ":1.0.0.0@" + checksum)
42+
require.NoError(t, err)
43+
44+
const workers = 8
45+
var wg sync.WaitGroup
46+
wg.Add(workers)
47+
errs := make(chan error, workers)
48+
for i := 0; i < workers; i++ {
49+
go func() {
50+
defer wg.Done()
51+
_, _, err := InstallPlugin(
52+
tenantID,
53+
identifier,
54+
plugin_entities.PLUGIN_RUNTIME_TYPE_LOCAL,
55+
&plugin_entities.PluginDeclaration{},
56+
"unittest",
57+
map[string]any{"from": "test"},
58+
)
59+
errs <- err
60+
}()
61+
}
62+
wg.Wait()
63+
close(errs)
64+
65+
// Validate DB state: exactly one plugin and one installation persisted
66+
plugins, err := db.GetAll[models.Plugin](
67+
db.Equal("plugin_unique_identifier", identifier.String()),
68+
db.Equal("install_type", string(plugin_entities.PLUGIN_RUNTIME_TYPE_LOCAL)),
69+
)
70+
require.NoError(t, err)
71+
require.Len(t, plugins, 1, "should persist exactly one plugin record")
72+
73+
installations, err := db.GetAll[models.PluginInstallation](
74+
db.Equal("plugin_unique_identifier", identifier.String()),
75+
db.Equal("tenant_id", tenantID),
76+
)
77+
require.NoError(t, err)
78+
require.Len(t, installations, 1, "should persist exactly one installation record for tenant")
79+
80+
// A subsequent sequential install should be rejected as already installed
81+
_, _, err = InstallPlugin(
82+
tenantID,
83+
identifier,
84+
plugin_entities.PLUGIN_RUNTIME_TYPE_LOCAL,
85+
&plugin_entities.PluginDeclaration{},
86+
"unittest",
87+
map[string]any{"from": "test"},
88+
)
89+
require.ErrorIs(t, err, ErrPluginAlreadyInstalled)
90+
}

‎internal/types/models/plugin.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,8 @@ import (
88
type Plugin struct {
99
Model
1010
// PluginUniqueIdentifier is a unique identifier for the plugin, it contains version and checksum
11-
PluginUniqueIdentifier string `json:"plugin_unique_identifier" gorm:"index;size:255"`
11+
// Enforce uniqueness to guarantee idempotency under concurrency
12+
PluginUniqueIdentifier string `json:"plugin_unique_identifier" gorm:"size:255;uniqueIndex:idx_plugin_unique_identifier"`
1213
// PluginID is the id of the plugin, only plugin name is considered
1314
PluginID string `json:"id" gorm:"index;size:255"`
1415
Refers int `json:"refers" gorm:"default:0"`

‎internal/types/models/trigger.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,4 +6,4 @@ type TriggerInstallation struct {
66
Provider string `json:"provider" gorm:"column:provider;size:127;index;not null"`
77
PluginUniqueIdentifier string `json:"plugin_unique_identifier" gorm:"index;size:255"`
88
PluginID string `json:"plugin_id" gorm:"index;size:255"`
9-
}
9+
}

0 commit comments

Comments
 (0)