Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 17 additions & 3 deletions backend/internal/app/runtime_plugin.go
Original file line number Diff line number Diff line change
Expand Up @@ -58,10 +58,24 @@ func providePluginSource(cfg *config.Config, store *postgres.Store, configuratio
// providePluginManager assembles verified storage, installation policy, and process ownership.
// The source is shared with host capabilities to avoid a manager/process dependency
// cycle. Construction leaves reconciliation to the registered application lifecycle hook.
func providePluginManager(cfg *config.Config, info BuildInfo, source *plugin.DevSource, configuration *plugin.Configuration, packages *provider.PackageFiles, processes *provider.Processes) *plugin.Manager {
func providePluginManager(
cfg *config.Config,
info BuildInfo,
source *plugin.DevSource,
configuration *plugin.Configuration,
packages *provider.PackageFiles,
processes *provider.Processes,
) (*plugin.Manager, error) {
data, err := provider.NewDataFiles(filepath.Join(cfg.Plugins.Dir, "data"))
if err != nil {
return nil, err
}
return plugin.NewManager(plugin.ManagerOptions{
Enabled: cfg.Plugins.Enabled, MaxRunning: cfg.Plugins.MaxRunning, HostVersion: info.Version, StateRoot: filepath.Join(cfg.Plugins.Dir, "state"),
}, source, packages, processes, configuration, slog.Default())
Enabled: cfg.Plugins.Enabled,
MaxRunning: cfg.Plugins.MaxRunning,
HostVersion: info.Version,
StateRoot: filepath.Join(cfg.Plugins.Dir, "state"),
}, source, packages, processes, configuration, data, slog.Default()), nil
}

// providePluginService binds administration to the same runtime manager and staged uploads.
Expand Down
4 changes: 4 additions & 0 deletions backend/internal/app/testdata/routes.json
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
"DELETE /api/v1/admin/payment/providers/:id",
"DELETE /api/v1/admin/plugin-indexes/:id",
"DELETE /api/v1/admin/plugins/:id",
"DELETE /api/v1/admin/plugins/:id/data",
"DELETE /api/v1/admin/pricing/configs/:id",
"DELETE /api/v1/admin/promo-codes/:id",
"DELETE /api/v1/admin/providers/:id",
Expand Down Expand Up @@ -156,8 +157,10 @@
"GET /api/v1/admin/plugins",
"GET /api/v1/admin/plugins/:id",
"GET /api/v1/admin/plugins/:id/config",
"GET /api/v1/admin/plugins/:id/data/archive",
"GET /api/v1/admin/plugins/:id/limits",
"GET /api/v1/admin/plugins/:id/logs",
"GET /api/v1/admin/plugins/data",
"GET /api/v1/admin/pricing/configs",
"GET /api/v1/admin/pricing/configs/:id",
"GET /api/v1/admin/pricing/defaults",
Expand Down Expand Up @@ -442,6 +445,7 @@
"POST /api/v1/admin/plugin-catalog/refresh",
"POST /api/v1/admin/plugin-indexes",
"POST /api/v1/admin/plugins",
"POST /api/v1/admin/plugins/:id/data/archive",
"POST /api/v1/admin/plugins/:id/disable",
"POST /api/v1/admin/plugins/:id/enable",
"POST /api/v1/admin/plugins/:id/restart",
Expand Down
5 changes: 4 additions & 1 deletion backend/internal/app/wire_gen.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

14 changes: 12 additions & 2 deletions backend/internal/pkg/locale/error_messages.json
Original file line number Diff line number Diff line change
Expand Up @@ -180,7 +180,12 @@
"MAX_RUNNING_EXCEEDED": "Plugin process limit reached",
"PLUGIN_INDEX_SIGNATURE_INVALID": "Plugin index signature is invalid",
"PLUGIN_CONFIG_INVALID": "Plugin configuration is invalid",
"PLUGIN_PACKAGE_MISSING": "Plugin package is missing; upload the same version again"
"PLUGIN_PACKAGE_MISSING": "Plugin package is missing; upload the same version again",
"PLUGIN_DATA_REQUIRES_STOP": "Disable the plugin before backing up or restoring its data.",
"PLUGIN_DATA_NOT_FOUND": "Plugin data not found.",
"PLUGIN_DATA_EXISTS": "Plugin data directory is not empty.",
"PLUGIN_DATA_INVALID": "Plugin data archive or directory is invalid.",
"PLUGIN_DATA_TOO_LARGE": "Plugin data exceeds archive limits."
},
"zh-Hans": {
"HTTP_400": "请求参数无效,请检查后重试。",
Expand Down Expand Up @@ -363,6 +368,11 @@
"MAX_RUNNING_EXCEEDED": "已达到插件进程数量上限。",
"PLUGIN_INDEX_SIGNATURE_INVALID": "插件索引签名无效。",
"PLUGIN_CONFIG_INVALID": "插件配置无效,请检查字段。",
"PLUGIN_PACKAGE_MISSING": "插件安装包丢失,请重新上传相同版本。"
"PLUGIN_PACKAGE_MISSING": "插件安装包丢失,请重新上传相同版本。",
"PLUGIN_DATA_REQUIRES_STOP": "备份或恢复数据前请先停用插件。",
"PLUGIN_DATA_NOT_FOUND": "插件数据不存在。",
"PLUGIN_DATA_EXISTS": "插件数据目录非空,请先备份并清理现有数据。",
"PLUGIN_DATA_INVALID": "插件数据备份或目录无效。",
"PLUGIN_DATA_TOO_LARGE": "插件数据超过备份大小或条目数量限制。"
}
}
169 changes: 169 additions & 0 deletions backend/internal/plugin/data.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,169 @@
package plugin

import (
"context"
"errors"
"io"
)

// data.go coordinates durable plugin directories with installation ownership.
// Files survive ordinary uninstall and remain discoverable by their plugin ID.
// Backup and restore require disabled or uninstalled plugins, and use the same
// per-plugin lock as installation changes so execution cannot restart midway.
//
// The file store owns directory validation and archive integrity. Management
// owns admission and callback draining before accessing stopped business data.
// Database KV and local files have separate commits: a failed file purge remains
// discoverable and can be retried after the installation row has been removed.
// External databases and object stores retain their operator-managed lifecycle.

const (
// DataMaxArchiveBytes bounds temporary disk use for compressed archives and expanded business files.
DataMaxArchiveBytes = int64(1 << 30)
)

// DataUsage describes local business files without exposing their absolute paths.
type DataUsage struct {
PluginID string `json:"plugin_id"` // publisher-qualified data owner
Bytes int64 `json:"bytes"` // regular file bytes
Files int64 `json:"files"` // regular file count
// Installed reports record presence; Enabled follows that record's desired
// execution state and stays false when the installation has been removed, even
// while local business files remain available for reinstallation.
Installed bool `json:"installed"` // installation record currently exists
Enabled bool `json:"enabled"` // installation currently requests execution
}

// DataArchive owns a completed immutable backup and its temporary file cleanup.
type DataArchive struct {
// Reader owns the immutable snapshot and its temporary file. Callers close
// it after delivery, including interrupted responses, to release that storage.
Reader io.ReadCloser
Bytes int64 // compressed archive byte length
}

// DataStorage manages plugin-owned directories and validated offline file archives.
// @project-doc docs/architecture/plugin_system.md#persistent_data
type DataStorage interface {
// Directory creates the stable directory for id and returns its absolute path.
// Callers serialize creation with purge and restore for this plugin. Existing
// data remains available across package replacements and ordinary uninstallation.
Directory(context.Context, string) (string, error)
// List measures retained directories without following links into other owners.
List(context.Context) ([]DataUsage, error)
// Remove deletes one owner's files after callers drain all plugin work.
// Missing directories succeed, allowing cleanup to resume after partial failure.
// Retained host financial and notification ledgers belong to their existing stores.
Remove(context.Context, string) error
// Backup snapshots stopped business files into a checksummed portable archive.
// Callers hold the plugin management lock and close the returned reader after
// use, releasing its temporary storage even when response delivery fails.
Backup(context.Context, string) (*DataArchive, error)
// Restore validates a complete archive before publishing an absent or empty directory.
// Nonempty destinations are rejected. Callers keep the plugin stopped and hold
// its management lock throughout validation and publication of restored files.
Restore(context.Context, string, io.Reader) error
}

// ListData returns durable directory usage including uninstalled plugin owners.
// Sizes are observations rather than quotas; running plugins may change them.
// Each record is annotated from current installation intent without reading secrets.
func (s *Service) ListData(ctx context.Context) ([]DataUsage, error) {
items, err := s.manager.data.List(ctx)
if err != nil {
return nil, err
}
installed, err := s.manager.store.List(ctx)
if err != nil {
return nil, err
}
byID := make(map[string]Installation, len(installed))
for _, item := range installed {
byID[item.PluginID] = item
}
for i := range items {
item, ok := byID[items[i].PluginID]
items[i].Installed = ok
items[i].Enabled = ok && item.Enabled
}
return items, nil
}

// PurgeData clears local files and KV for an uninstalled plugin identity.
// An installed owner must be removed first. The management lock prevents a
// concurrent reinstall from acquiring the directory while deletion is in progress.
//
// Host KV commits before local files are removed, because those stores cannot
// share a transaction. A filesystem failure leaves the owner discoverable for
// retry; successful cleanup also removes any package state left behind by an
// earlier partially completed uninstall. Financial and delivery ledger history remains.
func (s *Service) PurgeData(ctx context.Context, id string) error {
return s.manager.locked(ctx, id, func() error {
_, err := s.manager.store.Get(ctx, id)
if err == nil {
return ErrPluginAlreadyInstalled
}
if !errors.Is(err, ErrPluginNotFound) {
return err
}
if err := s.manager.drainLocked(ctx, id); err != nil {
return err
}
// Delete host KV before filesystem cleanup so failed file removal remains
// visible and retryable without allowing a new executable generation between steps.
if err := s.manager.store.Delete(ctx, id, true); err != nil {
return err
}
if err := s.manager.data.Remove(ctx, id); err != nil {
return err
}
return s.manager.syncLocked(ctx, id)
})
}

// BackupData produces a stable snapshot of a disabled or uninstalled owner's files.
// The returned archive is independent of later execution and belongs to the caller.
// Installed configuration, host KV, ledgers, and external stores require their respective backups.
//
// Stopped intent is checked while holding the same owner lock used by enable and
// reinstall, then actual process and callback completion is awaited. Only snapshot
// creation holds that lock; delivery reads an immutable temporary file, allowing
// later execution without making the downloaded bytes inconsistent.
func (s *Service) BackupData(ctx context.Context, id string) (*DataArchive, error) {
var archive *DataArchive
err := s.manager.locked(ctx, id, func() error {
if err := s.dataStopped(ctx, id); err != nil {
return err
}
var err error
archive, err = s.manager.data.Backup(ctx, id)
return err
})
return archive, err
}

// RestoreData publishes validated business files while keeping their owner stopped.
// The archive must match id and the destination must be absent or empty. Current
// installation configuration and permissions remain governed by ordinary administration.
func (s *Service) RestoreData(ctx context.Context, id string, reader io.Reader) error {
return s.manager.locked(ctx, id, func() error {
if err := s.dataStopped(ctx, id); err != nil {
return err
}
return s.manager.data.Restore(ctx, id, reader)
})
}

// dataStopped checks disabled intent and waits for actual executable callback completion.
// It runs under the owner's management lock. Drain failure leaves files untouched,
// and an enabled installation requires an administrator to disable it before retrying.
func (s *Service) dataStopped(ctx context.Context, id string) error {
item, err := s.manager.store.Get(ctx, id)
if err != nil && !errors.Is(err, ErrPluginNotFound) {
return err
}
if item != nil && item.Enabled {
return ErrDataRequiresStop
}
return s.manager.drainLocked(ctx, id)
}
97 changes: 97 additions & 0 deletions backend/internal/plugin/data_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,97 @@
package plugin

import (
"context"
"errors"
"io"
"strings"
"testing"

"github.com/stretchr/testify/require"
)

// data_test.go checks persistent file management against installation and process state.
// Storage callbacks are controlled independently from host KV, exposing ordering,
// retry, and stopped-owner decisions without starting an executable or external database.

// dataFixture supplies the management observations needed by persistent data tests.
type dataFixture struct {
DataStorage // methods unused by individual scenarios
removed []string // local owner purge attempts
removeError error // injected filesystem cleanup failure
backups int // accepted offline snapshot attempts
path string // persistent directory supplied to processes
}

// Directory returns the fixture's stable path for native process initialization.
func (f *dataFixture) Directory(context.Context, string) (string, error) {
return f.path, nil
}

// Remove records each cleanup attempt and returns the configured filesystem outcome.
func (f *dataFixture) Remove(_ context.Context, id string) error {
f.removed = append(f.removed, id)
return f.removeError
}

// Backup counts snapshots admitted after management has checked stopped-owner policy.
func (f *dataFixture) Backup(context.Context, string) (*DataArchive, error) {
f.backups++
return &DataArchive{Reader: io.NopCloser(strings.NewReader(""))}, nil
}

// TestDataDirectoryReachesNativeProcess starts one accepted native contribution.
// Its immutable process specification receives the durable path separately from disposable state, and ordinary uninstall retains it.
func TestDataDirectoryReachesNativeProcess(t *testing.T) {
manager, service, store, processes, _ := managerFixture(t, 1)
files := &dataFixture{path: "/fixture/persistent/test.plugin"}
manager.data = files
require.NoError(t, store.Save(t.Context(), managerInstallation(t, "test.plugin", true)))
require.NoError(t, manager.Sync(t.Context(), "test.plugin"))
require.Equal(t, files.path, processes.started[0].DataDir)
require.NotEqual(t, processes.started[0].StateDir, processes.started[0].DataDir)
require.NoError(t, service.Uninstall(t.Context(), "test.plugin", false))
require.Empty(t, files.removed)
}

// TestDataPurgeRetriesAfterCommittedUninstall injects a local deletion failure after host records commit.
// Retrying the explicit purge removes remaining files without resurrecting installation intent or changing the retained ledger policy.
func TestDataPurgeRetriesAfterCommittedUninstall(t *testing.T) {
manager, service, store, _, _ := managerFixture(t, 1)
files := &dataFixture{removeError: errors.New("fixture disk failure")}
manager.data = files
require.NoError(t, store.Save(t.Context(), managerInstallation(t, "test.plugin", false)))
require.ErrorIs(t, service.Uninstall(t.Context(), "test.plugin", true), files.removeError)
_, err := store.Get(t.Context(), "test.plugin")
require.ErrorIs(t, err, ErrPluginNotFound)
files.removeError = nil
require.NoError(t, service.PurgeData(t.Context(), "test.plugin"))
require.Equal(t, []string{"test.plugin", "test.plugin"}, files.removed)
require.Equal(t, []bool{true, true}, store.purge)
}

// TestDataBackupRequiresDisabledIntentAndCompletedDrain varies desired state and callback completion.
// Enabled or incompletely drained owners never reach snapshot creation; a stopped owner can produce a snapshot afterward.
func TestDataBackupRequiresDisabledIntentAndCompletedDrain(t *testing.T) {
manager, service, store, processes, _ := managerFixture(t, 1)
files := &dataFixture{}
manager.data = files
item := managerInstallation(t, "test.plugin", true)
require.NoError(t, store.Save(t.Context(), item))
_, err := service.BackupData(t.Context(), item.PluginID)
require.ErrorIs(t, err, ErrDataRequiresStop)
require.Zero(t, files.backups)
item.Enabled = false
require.NoError(t, store.Save(t.Context(), item))
processes.stopError = errors.New("fixture callback still active")
_, err = service.BackupData(t.Context(), item.PluginID)
require.ErrorIs(t, err, processes.stopError)
require.Zero(t, files.backups)
processes.stopError = nil
archive, err := service.BackupData(t.Context(), item.PluginID)
require.NoError(t, err)
require.NoError(t, archive.Reader.Close())
require.Equal(t, 1, files.backups)
require.ErrorIs(t, service.PurgeData(t.Context(), item.PluginID), ErrPluginAlreadyInstalled)
require.Empty(t, files.removed)
}
1 change: 1 addition & 0 deletions backend/internal/plugin/doc.go
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
// - catalog.go selects source-qualified releases; catalog_service.go, catalog_refresh.go, and catalog_indexes.go own cache and registration coordination.
// - configuration.go, limits.go, and dev_source.go normalize persistent policy and development records.
// - grant.go, ledger.go, wallet.go, redeem.go, subscriptions.go, and capabilities.go define caller authority, financial rules, and audit collaboration.
// - data.go coordinates persistent directories, retained-owner cleanup, and offline backup management.
// - http.go defines transport-independent HTTP contribution values.
// - tasks.go schedules declared jobs and retains in-memory execution results.
// - kv.go and kv_maintenance.go enforce isolated JSON storage and sampled quota maintenance.
Expand Down
5 changes: 5 additions & 0 deletions backend/internal/plugin/errors.go
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,11 @@ var (
ErrIndexNotFound = apperror.NotFound("PLUGIN_INDEX_NOT_FOUND", "Plugin index not found")
ErrSecretKeyMissing = apperror.BadRequest("PLUGIN_SECRET_KEY_REQUIRED", "Configure a persistent security.secret_encryption_key before saving plugin secrets")
ErrLimitsInvalid = apperror.BadRequest("PLUGIN_LIMITS_INVALID", "Plugin limits are invalid")
ErrDataRequiresStop = apperror.Conflict("PLUGIN_DATA_REQUIRES_STOP", "Disable the plugin before backing up or restoring its data")
ErrDataNotFound = apperror.NotFound("PLUGIN_DATA_NOT_FOUND", "Plugin data not found")
ErrDataExists = apperror.Conflict("PLUGIN_DATA_EXISTS", "Plugin data directory is not empty")
ErrDataInvalid = apperror.BadRequest("PLUGIN_DATA_INVALID", "Plugin data archive or directory is invalid")
ErrDataTooLarge = apperror.BadRequest("PLUGIN_DATA_TOO_LARGE", "Plugin data exceeds archive limits")
ErrMaxRunning = apperror.Conflict("MAX_RUNNING_EXCEEDED", "Plugin process limit reached")
ErrIndexSignatureInvalid = apperror.BadRequest("PLUGIN_INDEX_SIGNATURE_INVALID", "Plugin index signature is invalid")
ErrConfigInvalid = apperror.BadRequest("PLUGIN_CONFIG_INVALID", "Plugin configuration is invalid")
Expand Down
2 changes: 1 addition & 1 deletion backend/internal/plugin/helpers_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -278,7 +278,7 @@ func managerFixture(t *testing.T, maximum int) (*Manager, *Service, *memoryInsta
store := newMemoryInstallations()
processes := &processFixture{}
packages := &packageFixture{}
manager := NewManager(ManagerOptions{Enabled: true, MaxRunning: maximum, HostVersion: "0.5.4", StateRoot: t.TempDir()}, store, packages, processes, NewConfiguration(nil, false), slog.New(slog.NewTextHandler(io.Discard, nil)))
manager := NewManager(ManagerOptions{Enabled: true, MaxRunning: maximum, HostVersion: "0.5.4", StateRoot: t.TempDir()}, store, packages, processes, NewConfiguration(nil, false), nil, slog.New(slog.NewTextHandler(io.Discard, nil)))
return manager, NewService(manager, nil, nil, nil), store, processes, packages
}

Expand Down
Loading