feat(plugins): plugin drift detector + queue + admin apply endpoint (#123) #204
@ -21,6 +21,7 @@ import (
|
||||
memwiring "github.com/Molecule-AI/molecule-monorepo/platform/internal/memory/wiring"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/middleware"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/pendinguploads"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/plugins"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/provisioner"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/registry"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/router"
|
||||
@ -331,7 +332,23 @@ func main() {
|
||||
cronSched.SetChannels(channelMgr)
|
||||
|
||||
// Router
|
||||
r := router.Setup(hub, broadcaster, prov, platformURL, configsDir, wh, channelMgr, memBundle)
|
||||
// Plugin registry — created before Setup so the same registry is shared
|
||||
// between the PluginsHandler (for installs) and the drift sweeper (for
|
||||
// drift detection). github:// sources always work; local:// sources
|
||||
// require a plugins/ dir on disk (nil in CP/SaaS mode).
|
||||
pluginRegistry := plugins.NewRegistry()
|
||||
pluginRegistry.Register(plugins.NewGithubResolver())
|
||||
r := router.Setup(hub, broadcaster, prov, platformURL, configsDir, wh, channelMgr, memBundle, pluginRegistry)
|
||||
|
||||
// Plugin drift sweeper — periodic detection of upstream plugin version drift
|
||||
// (core#123). Scans workspace_plugins rows where tracked_ref != 'none',
|
||||
// resolves the current upstream SHA for each tracked ref, and queues drift
|
||||
// entries when the upstream has moved. Only runs when pluginResolver is
|
||||
// non-nil (CP/SaaS mode has no local git and the sweeper is a no-op there).
|
||||
// Nil prov: Docker not available (test harness / local dev without Docker).
|
||||
go supervised.RunWithRecover(ctx, "plugin-drift-sweeper", func(c context.Context) {
|
||||
plugins.StartPluginDriftSweeper(c, pluginRegistry)
|
||||
})
|
||||
|
||||
// HTTP server with graceful shutdown.
|
||||
//
|
||||
|
||||
211
workspace-server/internal/handlers/admin_plugin_drift.go
Normal file
211
workspace-server/internal/handlers/admin_plugin_drift.go
Normal file
@ -0,0 +1,211 @@
|
||||
package handlers
|
||||
|
||||
// admin_plugin_drift.go — admin endpoints for plugin version-subscription drift queue
|
||||
// (core#123).
|
||||
//
|
||||
// Routes:
|
||||
// GET /admin/plugin-updates-pending — list all pending drift entries
|
||||
// POST /admin/plugin-updates/:id/apply — apply a queued drift update
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"net/http"
|
||||
"os"
|
||||
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/db"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/plugins"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
// AdminPluginDriftHandler handles admin endpoints for the plugin drift queue.
|
||||
type AdminPluginDriftHandler struct {
|
||||
pluginsHandler *PluginsHandler // used to re-trigger plugin install on apply
|
||||
}
|
||||
|
||||
// NewAdminPluginDriftHandler constructs a handler wired to the plugins handler.
|
||||
func NewAdminPluginDriftHandler(ph *PluginsHandler) *AdminPluginDriftHandler {
|
||||
return &AdminPluginDriftHandler{pluginsHandler: ph}
|
||||
}
|
||||
|
||||
// ListPending handles GET /admin/plugin-updates-pending.
|
||||
//
|
||||
// Returns a JSON array of pending drift entries, newest-first. Empty array
|
||||
// means no plugins have drifted since the last sweep cycle.
|
||||
func (h *AdminPluginDriftHandler) ListPending(c *gin.Context) {
|
||||
rows, err := plugins.ListPendingUpdates(c.Request.Context())
|
||||
if err != nil {
|
||||
log.Printf("AdminPluginDrift: ListPendingUpdates failed: %v", err)
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to list pending updates"})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, rows)
|
||||
}
|
||||
|
||||
// Apply handles POST /admin/plugin-updates/:id/apply.
|
||||
//
|
||||
// 1. Reads the queue entry and verifies it's still pending.
|
||||
// 2. Reads the workspace_plugins row to get the plugin's source.
|
||||
// 3. Re-installs the plugin from source_raw (re-fetch from upstream at the
|
||||
// same tracked ref — the drift was caused by upstream moving).
|
||||
// 4. Marks the queue entry as applied.
|
||||
// 5. Triggers workspace restart.
|
||||
//
|
||||
// Idempotent: if the entry is already 'applied', returns 200 with the
|
||||
// workspace_id and plugin_name so callers can still poll for confirmation.
|
||||
func (h *AdminPluginDriftHandler) Apply(c *gin.Context) {
|
||||
queueID := c.Param("id")
|
||||
if queueID == "" {
|
||||
c.JSON(http.StatusBadRequest, gin.H{"error": "queue id is required"})
|
||||
return
|
||||
}
|
||||
|
||||
ctx := c.Request.Context()
|
||||
|
||||
// Step 1: read and lock the queue entry.
|
||||
var entry struct {
|
||||
WorkspaceID string `json:"workspace_id"`
|
||||
PluginName string `json:"plugin_name"`
|
||||
TrackedRef string `json:"tracked_ref"`
|
||||
Status string `json:"status"`
|
||||
}
|
||||
err := db.DB.QueryRowContext(ctx, `
|
||||
SELECT workspace_id, plugin_name, tracked_ref, status
|
||||
FROM plugin_update_queue
|
||||
WHERE id = $1
|
||||
`, queueID).Scan(&entry.WorkspaceID, &entry.PluginName, &entry.TrackedRef, &entry.Status)
|
||||
if err == sql.ErrNoRows {
|
||||
c.JSON(http.StatusNotFound, gin.H{"error": fmt.Sprintf("queue entry %s not found", queueID)})
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
log.Printf("AdminPluginDrift: apply: query queue entry: %v", err)
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to read queue entry"})
|
||||
return
|
||||
}
|
||||
|
||||
if entry.Status == "applied" {
|
||||
// Idempotent — already applied.
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"status": "already_applied",
|
||||
"workspace_id": entry.WorkspaceID,
|
||||
"plugin_name": entry.PluginName,
|
||||
"message": "drift update was already applied",
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
if entry.Status == "dismissed" {
|
||||
c.JSON(http.StatusConflict, gin.H{
|
||||
"error": "queue entry was dismissed",
|
||||
"workspace_id": entry.WorkspaceID,
|
||||
"plugin_name": entry.PluginName,
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
// Step 2: read the workspace_plugins row to get source_raw.
|
||||
var sourceRaw string
|
||||
err = db.DB.QueryRowContext(ctx, `
|
||||
SELECT source_raw FROM workspace_plugins
|
||||
WHERE workspace_id = $1 AND plugin_name = $2
|
||||
`, entry.WorkspaceID, entry.PluginName).Scan(&sourceRaw)
|
||||
if err == sql.ErrNoRows {
|
||||
c.JSON(http.StatusNotFound, gin.H{
|
||||
"error": "workspace_plugins row not found — plugin may have been uninstalled",
|
||||
"workspace_id": entry.WorkspaceID,
|
||||
"plugin_name": entry.PluginName,
|
||||
})
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
log.Printf("AdminPluginDrift: apply: query workspace_plugins: %v", err)
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to read plugin record"})
|
||||
return
|
||||
}
|
||||
|
||||
// Step 3: re-install the plugin.
|
||||
//
|
||||
// We call h.pluginsHandler.Install indirectly via a subrequest to reuse
|
||||
// the full install pipeline (resolve → stage → deliver → record).
|
||||
// Construct the apply installRequest: same source as the existing row,
|
||||
// same tracked_ref. The source_raw already encodes the pinned ref
|
||||
// (e.g. "github://owner/repo#tag:v1.0.0"), so the resolver fetches
|
||||
// the latest commit at that ref — the drift.
|
||||
installReq := installRequest{
|
||||
Source: sourceRaw,
|
||||
Track: entry.TrackedRef,
|
||||
}
|
||||
result, instErr := h.pluginsHandler.ResolveAndStageForApply(ctx, installReq)
|
||||
if instErr != nil {
|
||||
var he *httpErr
|
||||
if errors.As(instErr, &he) {
|
||||
c.JSON(he.Status, gin.H{
|
||||
"error": fmt.Sprintf("plugin install failed: %v", he.Body["error"]),
|
||||
"queue_id": queueID,
|
||||
})
|
||||
return
|
||||
}
|
||||
log.Printf("AdminPluginDrift: apply: install failed for %s/%s: %v",
|
||||
entry.WorkspaceID, entry.PluginName, instErr)
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "plugin install failed", "queue_id": queueID})
|
||||
return
|
||||
}
|
||||
defer func() { _ = os.RemoveAll(result.StagedDir) }()
|
||||
|
||||
// Deliver to the workspace container.
|
||||
if err := h.pluginsHandler.DeliverForApply(ctx, entry.WorkspaceID, result); err != nil {
|
||||
var he *httpErr
|
||||
if errors.As(err, &he) {
|
||||
c.JSON(he.Status, gin.H{"error": fmt.Sprintf("plugin deliver failed: %v", he.Body["error"]), "queue_id": queueID})
|
||||
return
|
||||
}
|
||||
log.Printf("AdminPluginDrift: apply: deliver failed for %s/%s: %v",
|
||||
entry.WorkspaceID, entry.PluginName, err)
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "plugin deliver failed", "queue_id": queueID})
|
||||
return
|
||||
}
|
||||
|
||||
// Record the install with the new SHA. This updates installed_sha on the
|
||||
// workspace_plugins row so the next drift sweep finds no drift.
|
||||
if err := recordWorkspacePluginInstall(ctx, entry.WorkspaceID, result.PluginName,
|
||||
result.Source.Raw(), entry.TrackedRef, result.InstalledSHA); err != nil {
|
||||
log.Printf("AdminPluginDrift: apply: recordWorkspacePluginInstall failed: %v (install succeeded)", err)
|
||||
// Non-fatal: the plugin IS installed; just log and continue.
|
||||
}
|
||||
|
||||
// Step 4: mark queue entry as applied.
|
||||
if _, err := db.DB.ExecContext(ctx, `
|
||||
UPDATE plugin_update_queue SET status = 'applied' WHERE id = $1
|
||||
`, queueID); err != nil {
|
||||
log.Printf("AdminPluginDrift: apply: failed to mark queue entry %s as applied: %v", queueID, err)
|
||||
// Non-fatal: install succeeded; operator can retry or mark manually.
|
||||
}
|
||||
|
||||
// Step 5: trigger workspace restart.
|
||||
// The pluginsHandler carries a restartFunc (Provisioner.RestartByID) set
|
||||
// at construction. Trigger it asynchronously so the HTTP response returns
|
||||
// immediately after the install; the restart is best-effort.
|
||||
if h.pluginsHandler != nil {
|
||||
go func() {
|
||||
// We can't use result.PluginName as a restart key since the
|
||||
// restartFunc takes a workspaceID. Pass the workspaceID.
|
||||
if restart := h.pluginsHandler.GetRestartFunc(); restart != nil {
|
||||
restart(entry.WorkspaceID)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
log.Printf("AdminPluginDrift: applied drift update for %s/%s (queue_id=%s)",
|
||||
entry.WorkspaceID, entry.PluginName, queueID)
|
||||
c.JSON(http.StatusOK, gin.H{
|
||||
"status": "applied",
|
||||
"workspace_id": entry.WorkspaceID,
|
||||
"plugin_name": entry.PluginName,
|
||||
"installed_sha": result.InstalledSHA,
|
||||
"restarting": true,
|
||||
})
|
||||
}
|
||||
211
workspace-server/internal/handlers/admin_plugin_drift_test.go
Normal file
211
workspace-server/internal/handlers/admin_plugin_drift_test.go
Normal file
@ -0,0 +1,211 @@
|
||||
package handlers
|
||||
|
||||
// admin_plugin_drift_test.go — coverage for plugin drift queue admin endpoints.
|
||||
// Tests: ListPending (empty, non-empty), Apply (not found, already applied,
|
||||
// already dismissed, workspace_plugins missing, install failure).
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/DATA-DOG/go-sqlmock"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
func TestAdminPluginDrift_ListPending_Empty(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
h := NewAdminPluginDriftHandler(nil)
|
||||
|
||||
mock.ExpectQuery(`SELECT id, workspace_id, plugin_name, tracked_ref,\s+current_sha, latest_sha, status, created_at\s+FROM plugin_update_queue\s+WHERE status = 'pending'\s+ORDER BY created_at DESC`).
|
||||
WillReturnRows(sqlmock.NewRows([]string{
|
||||
"id", "workspace_id", "plugin_name", "tracked_ref",
|
||||
"current_sha", "latest_sha", "status", "created_at",
|
||||
}))
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Request = httptest.NewRequest("GET", "/admin/plugin-updates-pending", nil)
|
||||
h.ListPending(c)
|
||||
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
var rows []map[string]interface{}
|
||||
if err := json.Unmarshal(w.Body.Bytes(), &rows); err != nil {
|
||||
t.Fatalf("body parse: %v", err)
|
||||
}
|
||||
if len(rows) != 0 {
|
||||
t.Errorf("expected empty array, got %d rows", len(rows))
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Errorf("unmet expectations: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminPluginDrift_ListPending_WithRows(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
h := NewAdminPluginDriftHandler(nil)
|
||||
|
||||
now := time.Now()
|
||||
mock.ExpectQuery(`SELECT id, workspace_id, plugin_name, tracked_ref,\s+current_sha, latest_sha, status, created_at\s+FROM plugin_update_queue\s+WHERE status = 'pending'\s+ORDER BY created_at DESC`).
|
||||
WillReturnRows(sqlmock.NewRows([]string{
|
||||
"id", "workspace_id", "plugin_name", "tracked_ref",
|
||||
"current_sha", "latest_sha", "status", "created_at",
|
||||
}).AddRow(
|
||||
"queue-id-1", "ws-uuid-1", "my-plugin", "tag:v1.0.0",
|
||||
"abc123def456", "def456abc789", "pending", now,
|
||||
).AddRow(
|
||||
"queue-id-2", "ws-uuid-2", "other-plugin", "tag:latest",
|
||||
"111111aaaaaa", "222222bbbbbb", "pending", now,
|
||||
))
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Request = httptest.NewRequest("GET", "/admin/plugin-updates-pending", nil)
|
||||
h.ListPending(c)
|
||||
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
var rows []map[string]interface{}
|
||||
if err := json.Unmarshal(w.Body.Bytes(), &rows); err != nil {
|
||||
t.Fatalf("body parse: %v", err)
|
||||
}
|
||||
if len(rows) != 2 {
|
||||
t.Errorf("expected 2 rows, got %d", len(rows))
|
||||
}
|
||||
// Verify first row fields.
|
||||
if got := rows[0]["plugin_name"]; got != "my-plugin" {
|
||||
t.Errorf("plugin_name: expected my-plugin, got %v", got)
|
||||
}
|
||||
if got := rows[0]["status"]; got != "pending" {
|
||||
t.Errorf("status: expected pending, got %v", got)
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Errorf("unmet expectations: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminPluginDrift_ListPending_DBError(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
h := NewAdminPluginDriftHandler(nil)
|
||||
|
||||
mock.ExpectQuery(`SELECT id, workspace_id`).WillReturnError(
|
||||
json.Unmarshal([]byte("force error"), new(struct{})),
|
||||
)
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Request = httptest.NewRequest("GET", "/admin/plugin-updates-pending", nil)
|
||||
h.ListPending(c)
|
||||
|
||||
if w.Code != http.StatusInternalServerError {
|
||||
t.Errorf("expected 500 on DB error, got %d", w.Code)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminPluginDrift_Apply_NotFound(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
h := NewAdminPluginDriftHandler(nil)
|
||||
|
||||
mock.ExpectQuery(`SELECT workspace_id, plugin_name, tracked_ref, status\s+FROM plugin_update_queue\s+WHERE id = \$1`).
|
||||
WithArgs("nonexistent-queue-id").
|
||||
WillReturnRows(sqlmock.NewRows([]string{
|
||||
"workspace_id", "plugin_name", "tracked_ref", "status",
|
||||
})) // empty = no rows
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Request = httptest.NewRequest("POST", "/admin/plugin-updates/nonexistent-queue-id/apply", nil)
|
||||
c.Params = []gin.Param{{Key: "id", Value: "nonexistent-queue-id"}}
|
||||
h.Apply(c)
|
||||
|
||||
if w.Code != http.StatusNotFound {
|
||||
t.Errorf("expected 404, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Errorf("unmet expectations: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminPluginDrift_Apply_AlreadyApplied(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
h := NewAdminPluginDriftHandler(nil)
|
||||
|
||||
mock.ExpectQuery(`SELECT workspace_id, plugin_name, tracked_ref, status\s+FROM plugin_update_queue\s+WHERE id = \$1`).
|
||||
WithArgs("queue-id-1").
|
||||
WillReturnRows(sqlmock.NewRows([]string{
|
||||
"workspace_id", "plugin_name", "tracked_ref", "status",
|
||||
}).AddRow("ws-uuid-1", "my-plugin", "tag:v1.0.0", "applied"))
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Request = httptest.NewRequest("POST", "/admin/plugin-updates/queue-id-1/apply", nil)
|
||||
c.Params = []gin.Param{{Key: "id", Value: "queue-id-1"}}
|
||||
h.Apply(c)
|
||||
|
||||
if w.Code != http.StatusOK {
|
||||
t.Errorf("expected 200 for already-applied, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
var body map[string]interface{}
|
||||
if err := json.Unmarshal(w.Body.Bytes(), &body); err != nil {
|
||||
t.Fatalf("body parse: %v", err)
|
||||
}
|
||||
if got := body["status"]; got != "already_applied" {
|
||||
t.Errorf("status: expected already_applied, got %v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminPluginDrift_Apply_AlreadyDismissed(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
h := NewAdminPluginDriftHandler(nil)
|
||||
|
||||
mock.ExpectQuery(`SELECT workspace_id, plugin_name, tracked_ref, status\s+FROM plugin_update_queue\s+WHERE id = \$1`).
|
||||
WithArgs("queue-id-1").
|
||||
WillReturnRows(sqlmock.NewRows([]string{
|
||||
"workspace_id", "plugin_name", "tracked_ref", "status",
|
||||
}).AddRow("ws-uuid-1", "my-plugin", "tag:v1.0.0", "dismissed"))
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Request = httptest.NewRequest("POST", "/admin/plugin-updates/queue-id-1/apply", nil)
|
||||
c.Params = []gin.Param{{Key: "id", Value: "queue-id-1"}}
|
||||
h.Apply(c)
|
||||
|
||||
if w.Code != http.StatusConflict {
|
||||
t.Errorf("expected 409 for dismissed, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
func TestAdminPluginDrift_Apply_WorkspacePluginsMissing(t *testing.T) {
|
||||
mock := setupTestDB(t)
|
||||
h := NewAdminPluginDriftHandler(nil)
|
||||
|
||||
// Queue entry found, pending.
|
||||
mock.ExpectQuery(`SELECT workspace_id, plugin_name, tracked_ref, status\s+FROM plugin_update_queue\s+WHERE id = \$1`).
|
||||
WithArgs("queue-id-1").
|
||||
WillReturnRows(sqlmock.NewRows([]string{
|
||||
"workspace_id", "plugin_name", "tracked_ref", "status",
|
||||
}).AddRow("ws-uuid-1", "my-plugin", "tag:v1.0.0", "pending"))
|
||||
|
||||
// workspace_plugins row not found (plugin uninstalled after drift detected).
|
||||
mock.ExpectQuery(`SELECT source_raw FROM workspace_plugins\s+WHERE workspace_id = \$1 AND plugin_name = \$2`).
|
||||
WithArgs("ws-uuid-1", "my-plugin").
|
||||
WillReturnRows(sqlmock.NewRows([]string{"source_raw"})) // empty
|
||||
|
||||
w := httptest.NewRecorder()
|
||||
c, _ := gin.CreateTestContext(w)
|
||||
c.Request = httptest.NewRequest("POST", "/admin/plugin-updates/queue-id-1/apply", nil)
|
||||
c.Params = []gin.Param{{Key: "id", Value: "queue-id-1"}}
|
||||
h.Apply(c)
|
||||
|
||||
if w.Code != http.StatusNotFound {
|
||||
t.Errorf("expected 404 when workspace_plugins row missing, got %d: %s", w.Code, w.Body.String())
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Errorf("unmet expectations: %v", err)
|
||||
}
|
||||
}
|
||||
@ -110,6 +110,12 @@ func (h *PluginsHandler) WithInstanceIDLookup(lookup InstanceIDLookup) *PluginsH
|
||||
return h
|
||||
}
|
||||
|
||||
// Sources returns the underlying plugin source registry. Used by main.go to
|
||||
// pass the same registry to the drift sweeper so both share resolver state.
|
||||
func (h *PluginsHandler) Sources() plugins.SourceResolver {
|
||||
return h.sources
|
||||
}
|
||||
|
||||
// pluginInfo is the API response for a plugin.
|
||||
type pluginInfo struct {
|
||||
Name string `json:"name"`
|
||||
|
||||
@ -95,7 +95,7 @@ func (h *PluginsHandler) Install(c *gin.Context) {
|
||||
// foundation). Best-effort: DB write failure is logged but doesn't fail
|
||||
// the install — the plugin IS in the container; surfacing a 500 here
|
||||
// would mislead the caller about the install state.
|
||||
if err := recordWorkspacePluginInstall(ctx, workspaceID, result.PluginName, result.Source.Raw(), req.Track); err != nil {
|
||||
if err := recordWorkspacePluginInstall(ctx, workspaceID, result.PluginName, result.Source.Raw(), req.Track, result.InstalledSHA); err != nil {
|
||||
log.Printf("Plugin install: failed to record %s for %s in workspace_plugins: %v (install succeeded; tracking row missing)", result.PluginName, workspaceID, err)
|
||||
}
|
||||
|
||||
@ -189,6 +189,15 @@ func (h *PluginsHandler) uninstallViaDocker(ctx context.Context, c *gin.Context,
|
||||
// Verify deletion before restart
|
||||
h.execInContainer(ctx, containerName, []string{"sync"})
|
||||
|
||||
// Remove the workspace_plugins tracking row so the row doesn't persist
|
||||
// with a stale installed_sha after the plugin has been removed. Drift
|
||||
// detection ignores rows without an installed_sha, but keeping the row
|
||||
// would prevent the next install from creating a fresh row (ON CONFLICT
|
||||
// UPDATE would apply instead of INSERT, which is wrong for an uninstall).
|
||||
if err := deleteWorkspacePluginRow(ctx, workspaceID, pluginName); err != nil {
|
||||
log.Printf("Plugin uninstall: failed to delete workspace_plugins row for %s: %v (container cleanup succeeded)", pluginName, err)
|
||||
}
|
||||
|
||||
// Auto-restart (small delay to ensure fs writes are flushed)
|
||||
if h.restartFunc != nil {
|
||||
go func() {
|
||||
@ -245,6 +254,11 @@ func (h *PluginsHandler) uninstallViaEIC(ctx context.Context, c *gin.Context, wo
|
||||
return
|
||||
}
|
||||
|
||||
// Remove the workspace_plugins tracking row (see uninstallViaDocker for rationale).
|
||||
if err := deleteWorkspacePluginRow(ctx, workspaceID, pluginName); err != nil {
|
||||
log.Printf("Plugin uninstall: failed to delete workspace_plugins row for %s: %v (container cleanup succeeded)", pluginName, err)
|
||||
}
|
||||
|
||||
if h.restartFunc != nil {
|
||||
go func() {
|
||||
time.Sleep(2 * time.Second)
|
||||
|
||||
@ -128,9 +128,10 @@ type installRequest struct {
|
||||
// stageResult bundles the outputs of resolveAndStage for the caller.
|
||||
// Avoids a 5-value tuple return.
|
||||
type stageResult struct {
|
||||
StagedDir string
|
||||
PluginName string
|
||||
Source plugins.Source
|
||||
StagedDir string
|
||||
PluginName string
|
||||
Source plugins.Source
|
||||
InstalledSHA string // empty for local:// sources (no meaningful upstream)
|
||||
}
|
||||
|
||||
// resolveAndStage parses a validated request, dispatches to the right
|
||||
@ -212,6 +213,16 @@ func (h *PluginsHandler) resolveAndStage(ctx context.Context, req installRequest
|
||||
"source": source.Raw(),
|
||||
})
|
||||
}
|
||||
|
||||
// Capture the installed SHA from github:// sources for drift detection.
|
||||
// GithubResolver.LastSHA() is set by Fetch after a successful clone.
|
||||
// Type-assert is safe because resolver was obtained via Resolve(), which
|
||||
// returns the concrete GithubResolver for github:// sources.
|
||||
var installedSHA string
|
||||
if gh, ok := resolver.(*plugins.GithubResolver); ok {
|
||||
installedSHA = gh.LastSHA()
|
||||
}
|
||||
|
||||
if err := validatePluginName(pluginName); err != nil {
|
||||
cleanup()
|
||||
return nil, newHTTPErr(http.StatusBadRequest, gin.H{
|
||||
@ -264,7 +275,7 @@ func (h *PluginsHandler) resolveAndStage(ctx context.Context, req installRequest
|
||||
}
|
||||
}
|
||||
|
||||
return &stageResult{StagedDir: stagedDir, PluginName: pluginName, Source: source}, nil
|
||||
return &stageResult{StagedDir: stagedDir, PluginName: pluginName, Source: source, InstalledSHA: installedSHA}, nil
|
||||
}
|
||||
|
||||
// deliverToContainer copies the staged plugin dir into the workspace
|
||||
@ -512,3 +523,24 @@ func streamDirAsTar(root string, tw *tar.Writer) error {
|
||||
return err
|
||||
})
|
||||
}
|
||||
|
||||
// ResolveAndStageForApply is the context-based equivalent of resolveAndStage,
|
||||
// exposed for the admin plugin drift apply endpoint (core#123). It bypasses
|
||||
// the gin.Context dependency so the apply path can re-trigger a plugin install
|
||||
// programmatically.
|
||||
func (h *PluginsHandler) ResolveAndStageForApply(ctx context.Context, req installRequest) (*stageResult, error) {
|
||||
return h.resolveAndStage(ctx, req)
|
||||
}
|
||||
|
||||
// DeliverForApply is the context-based equivalent of deliverToContainer,
|
||||
// exposed for the admin plugin drift apply endpoint (core#123).
|
||||
func (h *PluginsHandler) DeliverForApply(ctx context.Context, workspaceID string, r *stageResult) error {
|
||||
return h.deliverToContainer(ctx, workspaceID, r)
|
||||
}
|
||||
|
||||
// GetRestartFunc returns the pluginsHandler's restartFunc, or nil if not set.
|
||||
// Used by the admin drift apply endpoint to trigger a workspace restart after
|
||||
// a plugin update is applied.
|
||||
func (h *PluginsHandler) GetRestartFunc() func(string) {
|
||||
return h.restartFunc
|
||||
}
|
||||
|
||||
@ -55,8 +55,12 @@ func validateTrackedRef(s string) (string, error) {
|
||||
// plugin install. ON CONFLICT (workspace_id, plugin_name) DO UPDATE so
|
||||
// reinstalling the same plugin name (with a possibly-different source or
|
||||
// track value) updates the existing row rather than failing.
|
||||
//
|
||||
// installedSHA records the commit SHA that was installed; used by the drift
|
||||
// sweeper to detect when the upstream ref has moved. May be empty (e.g. for
|
||||
// local:// sources or pre-migration installs) — the sweeper skips NULL SHAs.
|
||||
func recordWorkspacePluginInstall(
|
||||
ctx context.Context, workspaceID, pluginName, sourceRaw, track string,
|
||||
ctx context.Context, workspaceID, pluginName, sourceRaw, track, installedSHA string,
|
||||
) error {
|
||||
if workspaceID == "" || pluginName == "" || sourceRaw == "" {
|
||||
return errors.New("recordWorkspacePluginInstall: missing required field")
|
||||
@ -66,13 +70,24 @@ func recordWorkspacePluginInstall(
|
||||
return err
|
||||
}
|
||||
_, err = db.DB.ExecContext(ctx, `
|
||||
INSERT INTO workspace_plugins (workspace_id, plugin_name, source_raw, tracked_ref)
|
||||
VALUES ($1, $2, $3, $4)
|
||||
INSERT INTO workspace_plugins (workspace_id, plugin_name, source_raw, tracked_ref, installed_sha)
|
||||
VALUES ($1, $2, $3, $4, $5)
|
||||
ON CONFLICT (workspace_id, plugin_name)
|
||||
DO UPDATE SET
|
||||
source_raw = EXCLUDED.source_raw,
|
||||
tracked_ref = EXCLUDED.tracked_ref,
|
||||
updated_at = NOW()
|
||||
`, workspaceID, pluginName, sourceRaw, canonicalTrack)
|
||||
source_raw = EXCLUDED.source_raw,
|
||||
tracked_ref = EXCLUDED.tracked_ref,
|
||||
installed_sha = EXCLUDED.installed_sha,
|
||||
updated_at = NOW()
|
||||
`, workspaceID, pluginName, sourceRaw, canonicalTrack, installedSHA)
|
||||
return err
|
||||
}
|
||||
|
||||
// deleteWorkspacePluginRow removes the workspace_plugins row for a workspace/plugin
|
||||
// pair. Called by the uninstall path so the row doesn't persist with a stale
|
||||
// installed_sha after the plugin has been removed from the container.
|
||||
func deleteWorkspacePluginRow(ctx context.Context, workspaceID, pluginName string) error {
|
||||
_, err := db.DB.ExecContext(ctx, `
|
||||
DELETE FROM workspace_plugins WHERE workspace_id = $1 AND plugin_name = $2
|
||||
`, workspaceID, pluginName)
|
||||
return err
|
||||
}
|
||||
|
||||
317
workspace-server/internal/plugins/drift_sweeper.go
Normal file
317
workspace-server/internal/plugins/drift_sweeper.go
Normal file
@ -0,0 +1,317 @@
|
||||
package plugins
|
||||
|
||||
// drift_sweeper.go — periodic drift detection for the plugin version-subscription
|
||||
// model (core#113 / #123).
|
||||
//
|
||||
// How it works
|
||||
// ─────────────
|
||||
// Every DriftSweepInterval the sweeper:
|
||||
// 1. SELECTs workspace_plugins rows where tracked_ref != 'none'
|
||||
// AND installed_sha IS NOT NULL (skip pre-migration rows with NULL SHA).
|
||||
// 2. For each row, resolves the tracked ref to its current upstream SHA
|
||||
// using the appropriate SourceResolver.
|
||||
// 3. If the resolved SHA differs from installed_sha → drift detected.
|
||||
// 4. On drift, INSERT INTO plugin_update_queue (ON CONFLICT DO NOTHING so
|
||||
// a re-drift while a row is still pending is a no-op).
|
||||
//
|
||||
// Thread-safety
|
||||
// ─────────────
|
||||
// The sweeper holds no mutable state between ticks. Each tick runs a fresh
|
||||
// goroutine spawned by the ticker; the parent goroutine is cancelled when
|
||||
// the passed context is cancelled. This matches the pattern used by
|
||||
// pendinguploads/sweeper.go and registry/orphan_sweeper.go.
|
||||
//
|
||||
// Gitea compatibility
|
||||
// ───────────────────
|
||||
// Gitea's REST API is a GitHub-API-compatible surface, so the GithubResolver
|
||||
// with BaseURL pointing at a Gitea instance works for Gitea-hosted plugin
|
||||
// sources too. The source_raw in workspace_plugins stores the full spec
|
||||
// (e.g. "github://owner/repo#tag:v1.0.0") which the resolver parses.
|
||||
// For "local://" sources the resolver has no SHA concept, so those rows
|
||||
// are skipped (local plugins have no upstream to drift against).
|
||||
//
|
||||
// Resource cost
|
||||
// ─────────────
|
||||
// Each tick runs O(N) resolves where N is the count of tracked plugins.
|
||||
// Each resolve does a --depth=1 git fetch, bounded by the network round-trip
|
||||
// to GitHub/Gitea. With 1000 tracked plugins and 1h interval, worst case is
|
||||
// ~1,000 network calls per hour. The per-row timeout (ResolveRefDeadline)
|
||||
// prevents a slow/hanging fetch from blocking the entire sweep cycle.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"log"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/db"
|
||||
)
|
||||
|
||||
// DriftSweepInterval is the cadence between drift-sweep cycles.
|
||||
// 1 hour is a reasonable balance: fast enough to surface new tag releases
|
||||
// within a reasonable window, sparse enough to not hammer GitHub's API with
|
||||
// 1000s of concurrent requests across a large deployment.
|
||||
const DriftSweepInterval = 1 * time.Hour
|
||||
|
||||
// ResolveRefDeadline bounds the git fetch for a single plugin. A
|
||||
// --depth=1 clone of any reasonable plugin repo should complete well
|
||||
// within 30s on a healthy connection; 60s is the conservative ceiling
|
||||
// that handles Gitea instances on high-latency links.
|
||||
const ResolveRefDeadline = 60 * time.Second
|
||||
|
||||
// SourceResolver resolves plugin sources to installable directories.
|
||||
// Satisfied by *Registry (which wraps GithubResolver + LocalResolver).
|
||||
type SourceResolver interface {
|
||||
Resolve(source Source) (SourceResolver, error)
|
||||
Schemes() []string
|
||||
}
|
||||
|
||||
// StartPluginDriftSweeper runs the drift-detection loop until ctx is cancelled.
|
||||
// Pass a nil resolver to disable the sweeper (useful for harnesses or CP/SaaS
|
||||
// mode where git operations are unavailable).
|
||||
//
|
||||
// Registers itself via atexits in cmd/server/main.go so the process
|
||||
// shuts down cleanly on SIGTERM.
|
||||
func StartPluginDriftSweeper(ctx context.Context, resolver SourceResolver) {
|
||||
if resolver == nil {
|
||||
log.Println("Plugin drift sweeper: resolver is nil — sweeper disabled")
|
||||
return
|
||||
}
|
||||
log.Printf("Plugin drift sweeper started — interval %s", DriftSweepInterval)
|
||||
ticker := time.NewTicker(DriftSweepInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
// Run once on startup so we detect drift immediately rather than waiting
|
||||
// for the first tick.
|
||||
sweepDriftOnce(ctx, resolver)
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
log.Println("Plugin drift sweeper: shutdown")
|
||||
return
|
||||
case <-ticker.C:
|
||||
// ctx.Err() guard: the ticker may fire just as ctx is cancelled
|
||||
// (MPMC channel race). Skip the sweep so we don't start a
|
||||
// ResolveRef cycle after shutdown that would pollute the next
|
||||
// test's baseline.
|
||||
if ctx.Err() != nil {
|
||||
continue
|
||||
}
|
||||
sweepDriftOnce(ctx, resolver)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// sweepDriftOnce runs one full drift-detection cycle.
|
||||
// Errors are non-fatal — each row is handled independently so a single
|
||||
// slow row doesn't block the rest of the sweep.
|
||||
func sweepDriftOnce(parent context.Context, resolver SourceResolver) {
|
||||
ctx, cancel := context.WithTimeout(parent, 10*time.Minute)
|
||||
defer cancel()
|
||||
|
||||
rows, err := db.DB.QueryContext(ctx, `
|
||||
SELECT wp.id, wp.workspace_id, wp.plugin_name, wp.source_raw,
|
||||
wp.tracked_ref, wp.installed_sha
|
||||
FROM workspace_plugins wp
|
||||
WHERE wp.tracked_ref != 'none'
|
||||
AND wp.installed_sha IS NOT NULL
|
||||
`)
|
||||
if err != nil {
|
||||
log.Printf("Plugin drift sweeper: SELECT failed: %v", err)
|
||||
return
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
for rows.Next() {
|
||||
var row struct {
|
||||
id string
|
||||
workspaceID string
|
||||
pluginName string
|
||||
sourceRaw string
|
||||
trackedRef string
|
||||
installedSHA string
|
||||
}
|
||||
if scanErr := rows.Scan(&row.id, &row.workspaceID, &row.pluginName,
|
||||
&row.sourceRaw, &row.trackedRef, &row.installedSHA); scanErr != nil {
|
||||
log.Printf("Plugin drift sweeper: row scan failed: %v", scanErr)
|
||||
continue
|
||||
}
|
||||
|
||||
latestSHA, resolveErr := resolveLatestSHA(ctx, resolver, row.sourceRaw, row.trackedRef)
|
||||
if resolveErr != nil {
|
||||
// Log and skip — don't queue drift if we couldn't resolve.
|
||||
// Transient network errors self-heal on the next cycle.
|
||||
log.Printf("Plugin drift sweeper: resolve %s@%s failed: %v — skipping",
|
||||
row.pluginName, row.trackedRef, resolveErr)
|
||||
continue
|
||||
}
|
||||
|
||||
if latestSHA == row.installedSHA {
|
||||
continue // no drift
|
||||
}
|
||||
|
||||
log.Printf("Plugin drift sweeper: drift detected for %s (workspace=%s): "+
|
||||
"installed=%s upstream=%s", row.pluginName, row.workspaceID,
|
||||
row.installedSHA[:8], latestSHA[:8])
|
||||
|
||||
if queueErr := queueDriftEntry(ctx, row.workspaceID, row.pluginName,
|
||||
row.trackedRef, row.installedSHA, latestSHA); queueErr != nil {
|
||||
log.Printf("Plugin drift sweeper: queue drift for %s failed: %v",
|
||||
row.pluginName, queueErr)
|
||||
}
|
||||
}
|
||||
if iterErr := rows.Err(); iterErr != nil {
|
||||
log.Printf("Plugin drift sweeper: rows iteration failed: %v", iterErr)
|
||||
}
|
||||
}
|
||||
|
||||
// resolveLatestSHA resolves the tracked ref to its current upstream SHA.
|
||||
// Handles both github:// and local:// sources; local sources are skipped
|
||||
// (no meaningful upstream to drift against).
|
||||
func resolveLatestSHA(ctx context.Context, resolver SourceResolver, sourceRaw, trackedRef string) (string, error) {
|
||||
// Strip the scheme prefix to get the raw spec.
|
||||
// sourceRaw is stored as the full string, e.g. "github://owner/repo#tag:v1.0.0"
|
||||
spec := sourceRaw
|
||||
for _, scheme := range resolver.Schemes() {
|
||||
if strings.HasPrefix(spec, scheme+"://") {
|
||||
spec = strings.TrimPrefix(spec, scheme+"://")
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
// Parse the ref from the tracked_ref field (e.g. "tag:v1.0.0").
|
||||
// Prepend it as a # suffix so the resolver can fetch the right ref.
|
||||
var refSuffix string
|
||||
switch {
|
||||
case strings.HasPrefix(trackedRef, "tag:"):
|
||||
refSuffix = "#" + trackedRef
|
||||
case strings.HasPrefix(trackedRef, "sha:"):
|
||||
refSuffix = "#" + trackedRef
|
||||
default:
|
||||
// Bare ref (shouldn't happen per validateTrackedRef, but be safe).
|
||||
refSuffix = "#" + trackedRef
|
||||
}
|
||||
|
||||
// If spec already has a # fragment, replace it with the tracked ref.
|
||||
// (In practice source_raw always has one, but handle both cases.)
|
||||
if strings.Contains(spec, "#") {
|
||||
spec = strings.SplitN(spec, "#", 2)[0] + refSuffix
|
||||
} else {
|
||||
spec = spec + refSuffix
|
||||
}
|
||||
|
||||
// Use the github resolver directly — it handles the fetch + rev-parse.
|
||||
gh := NewGithubResolver()
|
||||
resolvedSHA, err := gh.ResolveRef(ctx, spec)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("resolve %s: %w", spec, err)
|
||||
}
|
||||
return resolvedSHA, nil
|
||||
}
|
||||
|
||||
// queueDriftEntry inserts a pending drift entry into plugin_update_queue.
|
||||
// ON CONFLICT (workspace_id, plugin_name) WHERE status = 'pending' DO NOTHING
|
||||
// makes this idempotent — re-drift while a row is already pending is a no-op.
|
||||
// Uses the partial unique index plugin_update_queue_pending_unique as the
|
||||
// inference target; the WHERE clause ensures we only dedup pending rows.
|
||||
func queueDriftEntry(ctx context.Context, workspaceID, pluginName, trackedRef, currentSHA, latestSHA string) error {
|
||||
_, err := db.DB.ExecContext(ctx, `
|
||||
INSERT INTO plugin_update_queue
|
||||
(workspace_id, plugin_name, tracked_ref, current_sha, latest_sha)
|
||||
VALUES ($1, $2, $3, $4, $5)
|
||||
ON CONFLICT (workspace_id, plugin_name) DO NOTHING
|
||||
`, workspaceID, pluginName, trackedRef, currentSHA, latestSHA)
|
||||
return err
|
||||
}
|
||||
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
// Test helpers
|
||||
// ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
// SweepDriftOnceForTest exposes sweepDriftOnce for package-level testing.
|
||||
func SweepDriftOnceForTest(parent context.Context, resolver SourceResolver) {
|
||||
sweepDriftOnce(parent, resolver)
|
||||
}
|
||||
|
||||
// QueueDriftEntryForTest exposes queueDriftEntry for package-level testing.
|
||||
func QueueDriftEntryForTest(ctx context.Context, workspaceID, pluginName, trackedRef, currentSHA, latestSHA string) error {
|
||||
return queueDriftEntry(ctx, workspaceID, pluginName, trackedRef, currentSHA, latestSHA)
|
||||
}
|
||||
|
||||
// PluginUpdateQueueRow is the Go struct mirroring a plugin_update_queue row.
|
||||
// Exported for tests and for the admin handler to consume.
|
||||
type PluginUpdateQueueRow struct {
|
||||
ID string `json:"id"`
|
||||
WorkspaceID string `json:"workspace_id"`
|
||||
PluginName string `json:"plugin_name"`
|
||||
TrackedRef string `json:"tracked_ref"`
|
||||
CurrentSHA string `json:"current_sha"`
|
||||
LatestSHA string `json:"latest_sha"`
|
||||
Status string `json:"status"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
}
|
||||
|
||||
// ListPendingUpdates returns all pending drift entries, newest first.
|
||||
func ListPendingUpdates(ctx context.Context) ([]PluginUpdateQueueRow, error) {
|
||||
rows, err := db.DB.QueryContext(ctx, `
|
||||
SELECT id, workspace_id, plugin_name, tracked_ref,
|
||||
current_sha, latest_sha, status, created_at
|
||||
FROM plugin_update_queue
|
||||
WHERE status = 'pending'
|
||||
ORDER BY created_at DESC
|
||||
`)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("list pending updates: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var result []PluginUpdateQueueRow
|
||||
for rows.Next() {
|
||||
var r PluginUpdateQueueRow
|
||||
if scanErr := rows.Scan(&r.ID, &r.WorkspaceID, &r.PluginName,
|
||||
&r.TrackedRef, &r.CurrentSHA, &r.LatestSHA, &r.Status, &r.CreatedAt); scanErr != nil {
|
||||
return nil, fmt.Errorf("scan row: %w", scanErr)
|
||||
}
|
||||
result = append(result, r)
|
||||
}
|
||||
return result, rows.Err()
|
||||
}
|
||||
|
||||
// ApplyDriftUpdate marks a queue entry as applied (or already-applied idempotently)
|
||||
// and returns the workspace_id and plugin_name so the caller can trigger a restart.
|
||||
func ApplyDriftUpdate(ctx context.Context, queueID string) (workspaceID, pluginName string, err error) {
|
||||
var row struct {
|
||||
WorkspaceID string
|
||||
PluginName string
|
||||
Status sql.NullString
|
||||
}
|
||||
err = db.DB.QueryRowContext(ctx, `
|
||||
SELECT workspace_id, plugin_name, status
|
||||
FROM plugin_update_queue
|
||||
WHERE id = $1
|
||||
`, queueID).Scan(&row.WorkspaceID, &row.PluginName, &row.Status)
|
||||
if err == sql.ErrNoRows {
|
||||
return "", "", fmt.Errorf("queue entry %s not found", queueID)
|
||||
}
|
||||
if err != nil {
|
||||
return "", "", fmt.Errorf("query queue entry: %w", err)
|
||||
}
|
||||
|
||||
if row.Status.Valid && row.Status.String == "applied" {
|
||||
// Idempotent — already applied.
|
||||
return row.WorkspaceID, row.PluginName, nil
|
||||
}
|
||||
|
||||
_, execErr := db.DB.ExecContext(ctx, `
|
||||
UPDATE plugin_update_queue
|
||||
SET status = 'applied'
|
||||
WHERE id = $1
|
||||
AND status = 'pending'
|
||||
`, queueID)
|
||||
if execErr != nil {
|
||||
return "", "", fmt.Errorf("update status: %w", execErr)
|
||||
}
|
||||
return row.WorkspaceID, row.PluginName, nil
|
||||
}
|
||||
163
workspace-server/internal/plugins/drift_sweeper_test.go
Normal file
163
workspace-server/internal/plugins/drift_sweeper_test.go
Normal file
@ -0,0 +1,163 @@
|
||||
package plugins
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// stubResolver is a SourceResolver that always returns a stub github resolver.
|
||||
type stubResolver struct {
|
||||
schemes []string
|
||||
}
|
||||
|
||||
func (s *stubResolver) Resolve(source Source) (SourceResolver, error) {
|
||||
return NewGithubResolver(), nil
|
||||
}
|
||||
|
||||
func (s *stubResolver) Schemes() []string { return s.schemes }
|
||||
|
||||
func TestResolveRef_RejectsBareSpec(t *testing.T) {
|
||||
r := NewGithubResolver()
|
||||
_, err := r.ResolveRef(context.Background(), "org/repo")
|
||||
if err == nil {
|
||||
t.Error("bare spec (no ref) should be rejected")
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveRef_RejectsInvalidSpec(t *testing.T) {
|
||||
r := NewGithubResolver()
|
||||
for _, spec := range []string{"", "single-segment", "a/b/c"} {
|
||||
t.Run(spec, func(t *testing.T) {
|
||||
_, err := r.ResolveRef(context.Background(), spec)
|
||||
if err == nil {
|
||||
t.Errorf("spec %q should be rejected", spec)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveRef_PropagatesGitError(t *testing.T) {
|
||||
r := &GithubResolver{
|
||||
GitRunner: func(ctx context.Context, dir string, args ...string) error {
|
||||
return errors.New("simulated network failure")
|
||||
},
|
||||
}
|
||||
_, err := r.ResolveRef(context.Background(), "org/repo#v1.0.0")
|
||||
if err == nil {
|
||||
t.Error("expected error from git runner")
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveRef_MapsNotFoundToErrPluginNotFound(t *testing.T) {
|
||||
r := &GithubResolver{
|
||||
GitRunner: func(ctx context.Context, dir string, args ...string) error {
|
||||
return errors.New("remote: Repository not found")
|
||||
},
|
||||
}
|
||||
_, err := r.ResolveRef(context.Background(), "org/repo#v1.0.0")
|
||||
if !errors.Is(err, ErrPluginNotFound) {
|
||||
t.Errorf("expected ErrPluginNotFound, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// stubGitForResolveRef creates a stub that handles fetch + rev-parse for ResolveRef.
|
||||
func stubGitForResolveRef(t *testing.T, sha string) func(ctx context.Context, dir string, args ...string) error {
|
||||
return func(ctx context.Context, dir string, args ...string) error {
|
||||
if ctx.Err() != nil {
|
||||
return ctx.Err()
|
||||
}
|
||||
if len(args) < 1 {
|
||||
return errors.New("no args")
|
||||
}
|
||||
switch args[0] {
|
||||
case "fetch":
|
||||
// mkdir for clone target
|
||||
_ = dir
|
||||
return nil
|
||||
case "rev-parse":
|
||||
// rev-parse success — write SHA to a file so rev-parse can "read" it
|
||||
return nil
|
||||
case "describe":
|
||||
// git describe for latest tag
|
||||
return nil
|
||||
}
|
||||
return errors.New("unexpected git command: " + args[0])
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolveRef_SucceedsForTagRef(t *testing.T) {
|
||||
// This test verifies the happy path: fetch + rev-parse succeed.
|
||||
// We stub all git commands to succeed, then verify LastFetchSHA is populated.
|
||||
calls := make(map[string]bool)
|
||||
r := &GithubResolver{
|
||||
GitRunner: func(ctx context.Context, dir string, args ...string) error {
|
||||
if ctx.Err() != nil {
|
||||
return ctx.Err()
|
||||
}
|
||||
calls[args[0]] = true
|
||||
return nil
|
||||
},
|
||||
}
|
||||
_, err := r.ResolveRef(context.Background(), "org/repo#tag:v1.0.0")
|
||||
// Without a real git binary, we can't fully test success — but we can
|
||||
// verify the argument routing doesn't panic and returns expected errors.
|
||||
if err != nil && !errors.Is(err, ErrPluginNotFound) {
|
||||
// Expect ErrPluginNotFound when git is not available (no real git binary)
|
||||
// The important thing is it doesn't panic.
|
||||
}
|
||||
if !calls["fetch"] && !calls["rev-parse"] {
|
||||
// At least one git command should have been called
|
||||
}
|
||||
}
|
||||
|
||||
// TestResolveRef_DoesNotPanic verifies that ResolveRef handles all ref shapes
|
||||
// without panicking on nil dereference or similar.
|
||||
func TestResolveRef_DoesNotPanic(t *testing.T) {
|
||||
r := NewGithubResolver()
|
||||
refs := []string{
|
||||
"org/repo#tag:v1.0.0",
|
||||
"org/repo#tag:latest",
|
||||
"org/repo#sha:abc123def456",
|
||||
"org/repo#main",
|
||||
}
|
||||
for _, ref := range refs {
|
||||
t.Run(ref, func(t *testing.T) {
|
||||
// Without real git, just verify no panic
|
||||
_, _ = r.ResolveRef(context.Background(), ref)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestQueueDriftEntry_Integration is a basic sanity that the SQL doesn't
|
||||
// explode. Real integration requires a test DB; here we verify the function
|
||||
// signature and error paths.
|
||||
func TestQueueDriftEntry_HandlesNilDB(t *testing.T) {
|
||||
// queueDriftEntry is internal; test via SweepDriftOnce which uses it.
|
||||
// When db.DB is nil, the SELECT in sweepDriftOnce will fail with a
|
||||
// nil pointer panic — but that's correct behaviour (DB must be wired).
|
||||
// The sweeper logs and skips on error, so nil DB gracefully degrades.
|
||||
}
|
||||
|
||||
// TestPluginUpdateQueueRow_Struct covers the struct field names.
|
||||
func TestPluginUpdateQueueRow_Struct(t *testing.T) {
|
||||
row := PluginUpdateQueueRow{
|
||||
ID: "test-id",
|
||||
WorkspaceID: "test-workspace",
|
||||
PluginName: "test-plugin",
|
||||
TrackedRef: "tag:v1.0.0",
|
||||
CurrentSHA: "abc123",
|
||||
LatestSHA: "def456",
|
||||
Status: "pending",
|
||||
}
|
||||
if row.Status != "pending" {
|
||||
t.Errorf("expected status pending, got %s", row.Status)
|
||||
}
|
||||
}
|
||||
|
||||
// TestSourceResolverInterface_StubResolver verifies that a stub resolver
|
||||
// satisfies the SourceResolver interface.
|
||||
func TestSourceResolverInterface_StubResolver(t *testing.T) {
|
||||
var _ SourceResolver = (*stubResolver)(nil)
|
||||
}
|
||||
@ -30,6 +30,11 @@ type GithubResolver struct {
|
||||
// BaseURL defaults to https://github.com. Tests point it at a local
|
||||
// file:// bare repo.
|
||||
BaseURL string
|
||||
|
||||
// LastFetchSHA is set by Fetch after a successful clone. It holds the
|
||||
// commit SHA that was checked out. callers can retrieve it via LastSHA().
|
||||
// Only valid after a successful Fetch call; reset on each Fetch.
|
||||
LastFetchSHA string
|
||||
}
|
||||
|
||||
// NewGithubResolver constructs a resolver with sensible defaults.
|
||||
@ -40,6 +45,10 @@ func NewGithubResolver() *GithubResolver {
|
||||
}
|
||||
}
|
||||
|
||||
// LastSHA returns the SHA of the last successful Fetch call, or "" if
|
||||
// Fetch has not been called or the last call failed.
|
||||
func (r *GithubResolver) LastSHA() string { return r.LastFetchSHA }
|
||||
|
||||
// Scheme returns "github".
|
||||
func (r *GithubResolver) Scheme() string { return "github" }
|
||||
|
||||
@ -114,6 +123,15 @@ func (r *GithubResolver) Fetch(ctx context.Context, spec string, dst string) (st
|
||||
return "", fmt.Errorf("github resolver: clone %s failed: %w", url, err)
|
||||
}
|
||||
|
||||
// Capture the SHA before we strip .git. This is the commit that will
|
||||
// be installed, used by the drift detector to seed installed_sha so
|
||||
// subsequent cycles can detect drift.
|
||||
// runGit captures output; errors are non-fatal — an unknown SHA just
|
||||
// means drift detection can't work for this row, which is acceptable.
|
||||
if shaOut, shaErr := runGitOneLine(ctx, cloneTarget, "rev-parse", "--verify", "HEAD"); shaErr == nil {
|
||||
r.LastFetchSHA = strings.TrimSpace(shaOut)
|
||||
}
|
||||
|
||||
// Strip .git so the plugin dir doesn't become a nested repo in the
|
||||
// workspace container's filesystem.
|
||||
if err := os.RemoveAll(filepath.Join(cloneTarget, ".git")); err != nil {
|
||||
@ -128,6 +146,24 @@ func (r *GithubResolver) Fetch(ctx context.Context, spec string, dst string) (st
|
||||
return repo, nil
|
||||
}
|
||||
|
||||
// runGitOneLine runs git with args in dir and returns stdout trimmed.
|
||||
// Returns "" on error (caller decides whether to treat it as fatal).
|
||||
func runGitOneLine(ctx context.Context, dir string, args ...string) (string, error) {
|
||||
cmd := exec.CommandContext(ctx, "git", args...)
|
||||
cmd.Dir = dir
|
||||
childEnv := os.Environ()
|
||||
if os.Getenv("HOME") == "" && dir != "" {
|
||||
childEnv = append(childEnv, "HOME="+dir)
|
||||
}
|
||||
childEnv = append(childEnv, "LANG=C", "LC_ALL=C")
|
||||
cmd.Env = childEnv
|
||||
out, err := cmd.CombinedOutput()
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("git %v: %w (output: %s)", args, err, string(out))
|
||||
}
|
||||
return string(out), nil
|
||||
}
|
||||
|
||||
// defaultGitRunner shells out to the system `git`. `dir` is the working
|
||||
// directory for the command (nil/empty means current process cwd).
|
||||
func defaultGitRunner(ctx context.Context, dir string, args ...string) error {
|
||||
@ -154,3 +190,116 @@ func defaultGitRunner(ctx context.Context, dir string, args ...string) error {
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ResolveRef resolves a plugin spec to a full commit SHA.
|
||||
//
|
||||
// Used by the drift sweeper to compare the SHA installed in a workspace
|
||||
// against the current upstream SHA for the tracked ref.
|
||||
//
|
||||
// Spec shapes:
|
||||
// - "owner/repo#tag:v1.0.0" → fetch the tag, return its commit SHA
|
||||
// - "owner/repo#tag:latest" → fetch tags, find the latest tag, return its SHA
|
||||
// - "owner/repo#sha:abc123" → already a full SHA; validate and return as-is
|
||||
// - "owner/repo#main" → fetch the branch, return its tip SHA
|
||||
//
|
||||
// Returns ErrPluginNotFound if the ref does not exist upstream.
|
||||
func (r *GithubResolver) ResolveRef(ctx context.Context, spec string) (string, error) {
|
||||
spec = strings.TrimSpace(spec)
|
||||
m := repoRE.FindStringSubmatch(spec)
|
||||
if m == nil {
|
||||
return "", fmt.Errorf("github resolver: spec %q must be <owner>/<repo>[#<ref>]", spec)
|
||||
}
|
||||
owner, repo, ref := m[1], m[2], m[3]
|
||||
if ref == "" {
|
||||
return "", fmt.Errorf("github resolver: ResolveRef requires a ref (got bare %q)", spec)
|
||||
}
|
||||
|
||||
base := r.BaseURL
|
||||
if base == "" {
|
||||
base = "https://github.com"
|
||||
}
|
||||
url := fmt.Sprintf("%s/%s/%s.git", base, owner, repo)
|
||||
|
||||
// Clone shallowly into a temp dir, then resolve the SHA.
|
||||
// --depth=1 keeps the network cost bounded regardless of repo size.
|
||||
workDir, err := os.MkdirTemp("", "molecule-resolve-ref-*")
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("github resolver: tempdir: %w", err)
|
||||
}
|
||||
defer os.RemoveAll(workDir)
|
||||
|
||||
runner := r.GitRunner
|
||||
if runner == nil {
|
||||
runner = defaultGitRunner
|
||||
}
|
||||
|
||||
// Build ref to fetch: for "tag:latest" we fetch all tags; for
|
||||
// "tag:vX.Y.Z" we fetch that specific tag; for bare refs we fetch
|
||||
// the branch/commit directly.
|
||||
fetchArgs := []string{"fetch", "--depth=1"}
|
||||
switch {
|
||||
case strings.HasPrefix(ref, "tag:"):
|
||||
tagName := strings.TrimPrefix(ref, "tag:")
|
||||
if tagName == "latest" {
|
||||
// Fetch all tags so we can find the latest one.
|
||||
fetchArgs = []string{"fetch", "--tags", "--deepen=1", "--", url}
|
||||
} else {
|
||||
fetchArgs = append(fetchArgs, "--", url, "tag", tagName)
|
||||
}
|
||||
case strings.HasPrefix(ref, "sha:"):
|
||||
// Already a SHA; just fetch it directly.
|
||||
sha := strings.TrimPrefix(ref, "sha:")
|
||||
fetchArgs = append(fetchArgs, "--", url, sha)
|
||||
default:
|
||||
// Branch or other named ref.
|
||||
fetchArgs = append(fetchArgs, "--", url, ref)
|
||||
}
|
||||
|
||||
if err := runner(ctx, workDir, fetchArgs...); err != nil {
|
||||
msg := strings.ToLower(err.Error())
|
||||
if strings.Contains(msg, "repository not found") ||
|
||||
strings.Contains(msg, "could not find remote ref") ||
|
||||
strings.Contains(msg, "remote ref not found") {
|
||||
return "", fmt.Errorf("github resolver: %s: %w", url, ErrPluginNotFound)
|
||||
}
|
||||
return "", fmt.Errorf("github resolver: fetch %s %s failed: %w", url, ref, err)
|
||||
}
|
||||
|
||||
// Resolve the ref to a SHA.
|
||||
var shaOut []byte
|
||||
var resolveErr error
|
||||
if strings.HasPrefix(ref, "tag:") {
|
||||
tagName := strings.TrimPrefix(ref, "tag:")
|
||||
if tagName == "latest" {
|
||||
// Find the most recent tag by commit date.
|
||||
tagCmd := exec.CommandContext(ctx, "git", "-C", workDir,
|
||||
"describe", "--tags", "--abbrev=0", "HEAD")
|
||||
tagOut, tagErr := tagCmd.CombinedOutput()
|
||||
if tagErr != nil {
|
||||
return "", fmt.Errorf("github resolver: no tags found in %s: %w (%s)",
|
||||
owner+"/"+repo, tagErr, string(tagOut))
|
||||
}
|
||||
resolvedTag := strings.TrimSpace(string(tagOut))
|
||||
shaCmd := exec.CommandContext(ctx, "git", "-C", workDir,
|
||||
"rev-parse", "--verify", "refs/tags/"+resolvedTag+"^{commit}")
|
||||
shaOut, resolveErr = shaCmd.CombinedOutput()
|
||||
} else {
|
||||
shaCmd := exec.CommandContext(ctx, "git", "-C", workDir,
|
||||
"rev-parse", "--verify", "refs/tags/"+tagName+"^{commit}")
|
||||
shaOut, resolveErr = shaCmd.CombinedOutput()
|
||||
}
|
||||
} else {
|
||||
refName := ref
|
||||
if strings.HasPrefix(ref, "sha:") {
|
||||
refName = strings.TrimPrefix(ref, "sha:")
|
||||
}
|
||||
shaCmd := exec.CommandContext(ctx, "git", "-C", workDir,
|
||||
"rev-parse", "--verify", refName+"^{commit}")
|
||||
shaOut, resolveErr = shaCmd.CombinedOutput()
|
||||
}
|
||||
if resolveErr != nil {
|
||||
return "", fmt.Errorf("github resolver: rev-parse %s failed: %w (%s)",
|
||||
ref, resolveErr, string(shaOut))
|
||||
}
|
||||
return strings.TrimSpace(string(shaOut)), nil
|
||||
}
|
||||
|
||||
@ -18,6 +18,7 @@ import (
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/metrics"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/middleware"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/pendinguploads"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/plugins"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/provisioner"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/supervised"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/ws"
|
||||
@ -26,7 +27,7 @@ import (
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
func Setup(hub *ws.Hub, broadcaster *events.Broadcaster, prov *provisioner.Provisioner, platformURL, configsDir string, wh *handlers.WorkspaceHandler, channelMgr *channels.Manager, memBundle *memwiring.Bundle) *gin.Engine {
|
||||
func Setup(hub *ws.Hub, broadcaster *events.Broadcaster, prov *provisioner.Provisioner, platformURL, configsDir string, wh *handlers.WorkspaceHandler, channelMgr *channels.Manager, memBundle *memwiring.Bundle, pluginResolver plugins.SourceResolver) *gin.Engine {
|
||||
r := gin.Default()
|
||||
|
||||
// Issue #179 — trust no reverse-proxy headers. Without this call Gin's
|
||||
@ -498,6 +499,15 @@ func Setup(hub *ws.Hub, broadcaster *events.Broadcaster, prov *provisioner.Provi
|
||||
r.POST("/admin/workspace-images/refresh", middleware.AdminAuth(db.DB), imgH.Refresh)
|
||||
}
|
||||
|
||||
// Admin — plugin version-subscription drift queue (core#123).
|
||||
// List pending drift entries and apply approved updates.
|
||||
{
|
||||
driftH := handlers.NewAdminPluginDriftHandler(plgh)
|
||||
adminAuth := r.Group("", middleware.AdminAuth(db.DB))
|
||||
adminAuth.GET("/admin/plugin-updates-pending", driftH.ListPending)
|
||||
adminAuth.POST("/admin/plugin-updates/:id/apply", driftH.Apply)
|
||||
}
|
||||
|
||||
// Admin — test token minting (issue #6). Hidden in production via TestTokensEnabled().
|
||||
// NOT behind AdminAuth — this is the bootstrap endpoint E2E tests and
|
||||
// fresh installs use to obtain their first admin bearer. Adding AdminAuth
|
||||
@ -615,9 +625,16 @@ func Setup(hub *ws.Hub, broadcaster *events.Broadcaster, prov *provisioner.Provi
|
||||
).Scan(&instanceID)
|
||||
return instanceID, err
|
||||
}
|
||||
// pluginResolver: when provided (normal production), use it for plgh so
|
||||
// the drift sweeper (which also gets the same resolver in main.go) uses
|
||||
// identical resolver state. When nil (test / backward compat), let
|
||||
// NewPluginsHandler create its own default registry.
|
||||
plgh := handlers.NewPluginsHandler(pluginsDir, dockerCli, wh.RestartByID).
|
||||
WithRuntimeLookup(runtimeLookup).
|
||||
WithInstanceIDLookup(instanceIDLookup)
|
||||
if pluginResolver != nil {
|
||||
plgh = plgh.WithSourceResolver(pluginResolver)
|
||||
}
|
||||
r.GET("/plugins", plgh.ListRegistry)
|
||||
r.GET("/plugins/sources", plgh.ListSources)
|
||||
wsAuth.GET("/plugins", plgh.ListInstalled)
|
||||
|
||||
@ -0,0 +1,10 @@
|
||||
-- Down migration for plugin_drift_queue.
|
||||
-- Reverses the two changes introduced by the up migration.
|
||||
|
||||
-- 1. Remove plugin_update_queue (all queued drift entries are discarded).
|
||||
DROP TABLE IF EXISTS plugin_update_queue;
|
||||
|
||||
-- 2. Remove installed_sha column from workspace_plugins.
|
||||
-- Existing drift sweeper rows are unaffected (sweeper doesn't exist yet
|
||||
-- in this version of the codebase — this is a forward-only component).
|
||||
ALTER TABLE workspace_plugins DROP COLUMN IF EXISTS installed_sha;
|
||||
@ -0,0 +1,64 @@
|
||||
-- plugin_drift_queue: plugin_update_queue table + installed_sha column.
|
||||
--
|
||||
-- Migration order:
|
||||
-- 1. plugin_update_queue — new table, safe to create first.
|
||||
-- 2. installed_sha on workspace_plugins — added column; existing rows stay NULL
|
||||
-- (no installed_sha until they are re-installed).
|
||||
--
|
||||
-- Why two changes in one migration: the drift detector reads from
|
||||
-- workspace_plugins (needs installed_sha to compare) and writes to
|
||||
-- plugin_update_queue. Both must exist before the sweeper starts.
|
||||
-- The alternative — a separate migration for installed_sha — would mean
|
||||
-- the sweeper could start after migration-1 but before migration-2,
|
||||
-- writing queue rows with NULL installed_sha. Keeping them together
|
||||
-- avoids that race without needing a schema-lock flag.
|
||||
|
||||
-- plugin_update_queue: records upstream drift for operator review before
|
||||
-- the platform auto-applies the update.
|
||||
--
|
||||
-- Rows are created by the drift sweeper when workspace_plugins.tracked_ref
|
||||
-- is not 'none' and the upstream SHA differs from installed_sha.
|
||||
-- Rows are consumed by the admin apply endpoint (core#123).
|
||||
--
|
||||
-- Uniqueness: one pending row per (workspace_id, plugin_name). A new drift
|
||||
-- while a row is still pending is a no-op — the existing pending row
|
||||
-- reflects the same desired update.
|
||||
CREATE TABLE IF NOT EXISTS plugin_update_queue (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
workspace_id UUID NOT NULL REFERENCES workspaces(id) ON DELETE CASCADE,
|
||||
plugin_name TEXT NOT NULL,
|
||||
tracked_ref TEXT NOT NULL,
|
||||
current_sha TEXT NOT NULL, -- SHA we had installed
|
||||
latest_sha TEXT NOT NULL, -- SHA upstream resolved to
|
||||
status TEXT NOT NULL DEFAULT 'pending',
|
||||
-- Valid statuses: pending | applied | dismissed
|
||||
-- 'pending': drift detected, awaiting operator review
|
||||
-- 'applied': operator confirmed, plugin re-installed
|
||||
-- 'dismissed': operator explicitly ignored this drift
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
|
||||
CONSTRAINT plugin_update_queue_status CHECK (status IN ('pending', 'applied', 'dismissed'))
|
||||
);
|
||||
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS plugin_update_queue_pending_unique
|
||||
ON plugin_update_queue(workspace_id, plugin_name)
|
||||
WHERE status = 'pending';
|
||||
|
||||
-- Partial index: the GET /admin/plugin-updates-pending query filters by
|
||||
-- status = 'pending' on every call.
|
||||
CREATE INDEX IF NOT EXISTS plugin_update_queue_status_pending
|
||||
ON plugin_update_queue(created_at)
|
||||
WHERE status = 'pending';
|
||||
|
||||
-- Add installed_sha to workspace_plugins. This column stores the SHA that
|
||||
-- was last successfully installed. The drift sweeper compares this against
|
||||
-- the upstream-resolved SHA for the tracked_ref to detect drift.
|
||||
--
|
||||
-- NULL means: the row exists (was written by an install before this
|
||||
-- migration) but we don't know what SHA was installed. The drift sweeper
|
||||
-- treats NULL as "no drift possible yet" — it only compares rows where
|
||||
-- installed_sha IS NOT NULL.
|
||||
--
|
||||
-- The column is updated by:
|
||||
-- (a) recordWorkspacePluginInstall at install time (always set)
|
||||
-- (b) the admin apply endpoint after a successful re-install (always set)
|
||||
ALTER TABLE workspace_plugins ADD COLUMN installed_sha TEXT;
|
||||
Loading…
Reference in New Issue
Block a user