feat(deploy): 在线更新 - 新增后台检查、强制备份迁移与失败自动回退
This commit is contained in:
ddCat
@@ -3,16 +3,24 @@
|
||||
FROM golang:1.25-alpine AS backend-build
|
||||
|
||||
ARG GOPROXY=https://proxy.golang.org,direct
|
||||
ARG BUILD_VERSION
|
||||
ARG BUILD_COMMIT=unknown
|
||||
ARG BUILD_TIME=unknown
|
||||
ENV GOPROXY=$GOPROXY
|
||||
|
||||
RUN apk add --no-cache build-base
|
||||
|
||||
WORKDIR /src/backend
|
||||
COPY VERSION /src/VERSION
|
||||
COPY backend/go.mod backend/go.sum ./
|
||||
RUN --mount=type=cache,target=/go/pkg/mod go mod download
|
||||
COPY backend ./
|
||||
RUN --mount=type=cache,target=/root/.cache/go-build CGO_ENABLED=1 GOOS=linux go build -trimpath -ldflags="-s -w" -o /out/infinite-canvas-backend ./cmd/server \
|
||||
&& CGO_ENABLED=1 GOOS=linux go build -trimpath -ldflags="-s -w" -o /out/migrate-sqlite-postgres ./cmd/migrate-sqlite-postgres
|
||||
RUN --mount=type=cache,target=/root/.cache/go-build set -eu; \
|
||||
version="${BUILD_VERSION:-$(tr -d '\r\n' </src/VERSION)}"; \
|
||||
ldflags="-s -w -X infinite-canvas/backend/internal/buildinfo.Version=${version} -X infinite-canvas/backend/internal/buildinfo.Commit=${BUILD_COMMIT} -X infinite-canvas/backend/internal/buildinfo.BuildTime=${BUILD_TIME}"; \
|
||||
CGO_ENABLED=1 GOOS=linux go build -trimpath -ldflags="$ldflags" -o /out/infinite-canvas-backend ./cmd/server; \
|
||||
CGO_ENABLED=1 GOOS=linux go build -trimpath -ldflags="$ldflags" -o /out/migrate-schema ./cmd/migrate-schema; \
|
||||
CGO_ENABLED=1 GOOS=linux go build -trimpath -ldflags="$ldflags" -o /out/migrate-sqlite-postgres ./cmd/migrate-sqlite-postgres
|
||||
|
||||
FROM alpine:3.22
|
||||
|
||||
@@ -27,9 +35,10 @@ ENV CANVAS_BACKEND_DATA_DIR=/data
|
||||
|
||||
RUN mkdir -p /data && chown -R app:app /data
|
||||
COPY --from=backend-build /out/infinite-canvas-backend /usr/local/bin/infinite-canvas-backend
|
||||
COPY --from=backend-build /out/migrate-schema /usr/local/bin/migrate-schema
|
||||
COPY --from=backend-build /out/migrate-sqlite-postgres /usr/local/bin/migrate-sqlite-postgres
|
||||
|
||||
USER app
|
||||
EXPOSE 8080
|
||||
HEALTHCHECK --interval=30s --timeout=3s --start-period=10s --retries=3 CMD wget -qO- http://127.0.0.1:8080/api/health >/dev/null || exit 1
|
||||
HEALTHCHECK --interval=30s --timeout=3s --start-period=10s --retries=3 CMD wget -qO- http://127.0.0.1:8080/api/health/ready >/dev/null || exit 1
|
||||
CMD ["infinite-canvas-backend"]
|
||||
|
||||
131
backend/cmd/host-updater/main.go
Normal file
131
backend/cmd/host-updater/main.go
Normal file
@@ -0,0 +1,131 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"infinite-canvas/backend/internal/hostupdate"
|
||||
)
|
||||
|
||||
func main() {
|
||||
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
||||
defer stop()
|
||||
if err := run(ctx); err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func run(ctx context.Context) error {
|
||||
socketPath := env("CANVAS_UPDATER_SOCKET", "/run/open-ai-canvas-updater/updater.sock")
|
||||
token := strings.TrimSpace(os.Getenv("CANVAS_UPDATER_TOKEN"))
|
||||
manager, err := hostupdate.NewManager(hostupdate.Config{
|
||||
Repository: env("CANVAS_UPDATER_REPOSITORY", "ddcat-ai/open-ai-canvas"),
|
||||
InstallDir: env("CANVAS_UPDATER_INSTALL_DIR", "/opt/open-ai-canvas"),
|
||||
ComposeFile: env("CANVAS_UPDATER_COMPOSE_FILE", "docker-compose.deploy.yml"),
|
||||
EnvFile: env("CANVAS_UPDATER_ENV_FILE", ".env"),
|
||||
StateDir: env("CANVAS_UPDATER_STATE_DIR", "/var/lib/open-ai-canvas-updater"),
|
||||
BackupDir: env("CANVAS_UPDATER_BACKUP_DIR", "/opt/open-ai-canvas/backups"),
|
||||
HealthURL: strings.TrimSpace(os.Getenv("CANVAS_UPDATER_HEALTH_URL")),
|
||||
GitHubToken: strings.TrimSpace(os.Getenv("CANVAS_UPDATER_GITHUB_TOKEN")),
|
||||
StableWindow: envDuration("CANVAS_UPDATER_STABLE_WINDOW", 30*time.Second),
|
||||
StepTimeout: envDuration("CANVAS_UPDATER_STEP_TIMEOUT", 20*time.Minute),
|
||||
BinaryPath: env("CANVAS_UPDATER_BINARY_PATH", "/usr/local/bin/open-ai-canvas-host-updater"),
|
||||
ServiceName: env("CANVAS_UPDATER_SERVICE_NAME", "open-ai-canvas-updater.service"),
|
||||
SelfUpdate: envBool("CANVAS_UPDATER_SELF_UPDATE", true),
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
server, err := hostupdate.NewServer(manager, token)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := os.MkdirAll(filepath.Dir(socketPath), 0o755); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := removeStaleSocket(socketPath); err != nil {
|
||||
return err
|
||||
}
|
||||
listener, err := net.Listen("unix", socketPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer listener.Close()
|
||||
defer os.Remove(socketPath)
|
||||
if err := os.Chmod(socketPath, 0o666); err != nil {
|
||||
return err
|
||||
}
|
||||
httpServer := &http.Server{Handler: server.Handler(), ReadHeaderTimeout: 5 * time.Second}
|
||||
serveErr := make(chan error, 1)
|
||||
go func() { serveErr <- httpServer.Serve(listener) }()
|
||||
log.Printf("host updater listening on unix://%s", socketPath)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
return httpServer.Shutdown(shutdownCtx)
|
||||
case err := <-serveErr:
|
||||
if errors.Is(err, http.ErrServerClosed) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
func removeStaleSocket(path string) error {
|
||||
info, err := os.Lstat(path)
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if info.Mode()&os.ModeSocket == 0 {
|
||||
return fmt.Errorf("拒绝删除非 Socket 路径 %s", path)
|
||||
}
|
||||
return os.Remove(path)
|
||||
}
|
||||
|
||||
func env(key, fallback string) string {
|
||||
if value := strings.TrimSpace(os.Getenv(key)); value != "" {
|
||||
return value
|
||||
}
|
||||
return fallback
|
||||
}
|
||||
|
||||
func envDuration(key string, fallback time.Duration) time.Duration {
|
||||
value := strings.TrimSpace(os.Getenv(key))
|
||||
if value == "" {
|
||||
return fallback
|
||||
}
|
||||
parsed, err := time.ParseDuration(value)
|
||||
if err != nil || parsed <= 0 {
|
||||
log.Fatalf("%s 必须是正数时长", key)
|
||||
}
|
||||
return parsed
|
||||
}
|
||||
|
||||
func envBool(key string, fallback bool) bool {
|
||||
value := strings.ToLower(strings.TrimSpace(os.Getenv(key)))
|
||||
if value == "" {
|
||||
return fallback
|
||||
}
|
||||
if value == "true" || value == "1" || value == "yes" {
|
||||
return true
|
||||
}
|
||||
if value == "false" || value == "0" || value == "no" {
|
||||
return false
|
||||
}
|
||||
log.Fatalf("%s 必须是 true 或 false", key)
|
||||
return fallback
|
||||
}
|
||||
64
backend/cmd/migrate-schema/main.go
Normal file
64
backend/cmd/migrate-schema/main.go
Normal file
@@ -0,0 +1,64 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"strings"
|
||||
|
||||
"infinite-canvas/backend/internal/database"
|
||||
)
|
||||
|
||||
func main() {
|
||||
command := "up"
|
||||
if len(os.Args) > 1 {
|
||||
command = strings.ToLower(strings.TrimSpace(os.Args[1]))
|
||||
}
|
||||
db, err := database.Open(database.Config{
|
||||
Driver: env("CANVAS_DATABASE_DRIVER", "sqlite"),
|
||||
DSN: os.Getenv("DATABASE_URL"),
|
||||
DataDir: env("CANVAS_BACKEND_DATA_DIR", "data"),
|
||||
})
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
if err := database.ConfigurePool(db); err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
sqlDB, err := db.DB()
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
defer sqlDB.Close()
|
||||
|
||||
switch command {
|
||||
case "up":
|
||||
if err := database.MigrateSchema(db); err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
case "status":
|
||||
case "verify":
|
||||
if err := database.RequireSchemaVersion(db); err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
default:
|
||||
log.Fatalf("未知命令 %q；可用命令：up、status、verify", command)
|
||||
}
|
||||
status, err := database.ReadSchemaStatus(db)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
encoded, err := json.Marshal(status)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
fmt.Println(string(encoded))
|
||||
}
|
||||
|
||||
func env(key string, fallback string) string {
|
||||
if value := strings.TrimSpace(os.Getenv(key)); value != "" {
|
||||
return value
|
||||
}
|
||||
return fallback
|
||||
}
|
||||
@@ -1,26 +1,41 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strconv"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"infinite-canvas/backend/internal/database"
|
||||
"infinite-canvas/backend/internal/handler"
|
||||
"infinite-canvas/backend/internal/repository"
|
||||
"infinite-canvas/backend/internal/service"
|
||||
"infinite-canvas/backend/internal/updaterclient"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
func main() {
|
||||
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
||||
defer stop()
|
||||
if err := run(ctx); err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func run(ctx context.Context) error {
|
||||
dataDir := env("CANVAS_BACKEND_DATA_DIR", "data")
|
||||
if err := os.MkdirAll(dataDir, 0o755); err != nil {
|
||||
log.Fatal(err)
|
||||
return err
|
||||
}
|
||||
db, err := database.Open(database.Config{
|
||||
Driver: env("CANVAS_DATABASE_DRIVER", "sqlite"),
|
||||
@@ -28,44 +43,60 @@ func main() {
|
||||
DataDir: dataDir,
|
||||
})
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
return err
|
||||
}
|
||||
if err := database.ConfigurePool(db); err != nil {
|
||||
log.Fatal(err)
|
||||
return err
|
||||
}
|
||||
if err := database.MigrateSchema(db); err != nil {
|
||||
log.Fatal(err)
|
||||
sqlDB, err := db.DB()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer sqlDB.Close()
|
||||
autoMigrate, err := envBool("CANVAS_AUTO_MIGRATE", true)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if autoMigrate {
|
||||
err = database.MigrateSchema(db)
|
||||
} else {
|
||||
err = database.RequireSchemaVersion(db)
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
repo := repository.New(db)
|
||||
addr := env("CANVAS_BACKEND_ADDR", ":8080")
|
||||
capabilities := service.RuntimeCapabilitiesForDeployment(addr, os.Getenv("CANVAS_DESKTOP_LOCAL_CHANNELS_ENABLED"))
|
||||
svc := service.NewWithRuntimeCapabilities(repo, dataDir, capabilities)
|
||||
if updaterToken := strings.TrimSpace(os.Getenv("CANVAS_UPDATER_TOKEN")); updaterToken != "" {
|
||||
svc.ConfigureUpdateManager(updaterclient.New(env("CANVAS_UPDATER_SOCKET", "/run/open-ai-canvas-updater/updater.sock"), updaterToken))
|
||||
}
|
||||
defer svc.Close()
|
||||
if err := svc.ValidateRuntime(); err != nil {
|
||||
log.Fatal(err)
|
||||
return err
|
||||
}
|
||||
if err := svc.EnsureSystemChannelModels(); err != nil {
|
||||
log.Fatal(err)
|
||||
return err
|
||||
}
|
||||
if err := svc.EnsureDefaultPromptTemplates(); err != nil {
|
||||
log.Fatal(err)
|
||||
return err
|
||||
}
|
||||
if err := svc.EnsureBuiltinProjectWorkflowTemplate(); err != nil {
|
||||
log.Fatal(err)
|
||||
return err
|
||||
}
|
||||
if err := svc.EnsureBuiltinSkills(); err != nil {
|
||||
log.Fatal(err)
|
||||
return err
|
||||
}
|
||||
if err := svc.EnsureSkillPackages(); err != nil {
|
||||
log.Fatal(err)
|
||||
return err
|
||||
}
|
||||
if summary, err := svc.MigrateLegacyStorage(); err != nil {
|
||||
log.Printf("storage migration skipped after error: %v", err)
|
||||
} else if summary.Backup != "" {
|
||||
log.Printf("storage migration completed: tasks=%d assets=%d projects=%d backup=%s", summary.Tasks, summary.Assets, summary.Projects, summary.Backup)
|
||||
}
|
||||
svc.StartWorker()
|
||||
|
||||
r := gin.New()
|
||||
r.Use(gin.LoggerWithFormatter(func(param gin.LogFormatterParams) string {
|
||||
return fmt.Sprintf("%s - [%s] \"%s %s\" %d %s %s\n", param.ClientIP, param.TimeStamp.Format(time.RFC3339), param.Method, redactCanvasSharePath(param.Path), param.StatusCode, param.Latency, param.ErrorMessage)
|
||||
@@ -73,14 +104,13 @@ func main() {
|
||||
r.Use(handler.RequestCorrelationMiddleware())
|
||||
corsMiddleware, err := cors()
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
return err
|
||||
}
|
||||
r.Use(corsMiddleware)
|
||||
handler.ConfigureRuntime(svc)
|
||||
api := r.Group("/api")
|
||||
api.GET("/health", func(c *gin.Context) {
|
||||
c.JSON(200, gin.H{"code": 0, "data": gin.H{"status": "ok"}, "msg": "ok"})
|
||||
})
|
||||
status := newSystemStatus(db, svc)
|
||||
registerSystemStatusRoutes(api, status)
|
||||
handler.RegisterOAuthCallbackRoutes(r, svc)
|
||||
handler.RegisterAuthRoutes(api, svc)
|
||||
handler.RegisterFeatureAvailabilityRoutes(api, svc)
|
||||
@@ -88,6 +118,7 @@ func main() {
|
||||
handler.RegisterAdminRoutes(api, svc)
|
||||
handler.RegisterAdminAnalyticsRoutes(api, svc)
|
||||
handler.RegisterAdminStorageRoutes(api, svc)
|
||||
handler.RegisterAdminUpdateRoutes(api, svc)
|
||||
handler.RegisterAnnouncementRoutes(api, svc)
|
||||
handler.RegisterFinanceRoutes(api, svc)
|
||||
handler.RegisterLibTVRoutes(api, svc)
|
||||
@@ -112,10 +143,52 @@ func main() {
|
||||
handler.RegisterCanvasShareRoutes(api, svc)
|
||||
r.NoRoute(handler.SystemProxyNoRouteHandler(svc))
|
||||
|
||||
log.Printf("影策 backend listening on %s", addr)
|
||||
if err := r.Run(addr); err != nil {
|
||||
log.Fatal(err)
|
||||
listener, err := net.Listen("tcp", addr)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
workerTimeout, err := envDuration("CANVAS_SHUTDOWN_TIMEOUT", 10*time.Minute)
|
||||
if err != nil {
|
||||
_ = listener.Close()
|
||||
return err
|
||||
}
|
||||
httpServer := &http.Server{Handler: r, ReadHeaderTimeout: 10 * time.Second}
|
||||
svc.StartWorker()
|
||||
status.markStarted()
|
||||
serveErr := make(chan error, 1)
|
||||
go func() { serveErr <- httpServer.Serve(listener) }()
|
||||
log.Printf("影策 backend listening on %s", addr)
|
||||
|
||||
var serveFailure error
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case err := <-serveErr:
|
||||
if err != nil && !errors.Is(err, http.ErrServerClosed) {
|
||||
serveFailure = fmt.Errorf("HTTP 服务异常退出：%w", err)
|
||||
}
|
||||
}
|
||||
|
||||
status.beginDrain()
|
||||
var shutdownFailures []error
|
||||
if serveFailure != nil {
|
||||
shutdownFailures = append(shutdownFailures, serveFailure)
|
||||
}
|
||||
httpShutdownCtx, httpShutdownCancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer httpShutdownCancel()
|
||||
if err := httpServer.Shutdown(httpShutdownCtx); err != nil {
|
||||
_ = httpServer.Close()
|
||||
shutdownFailures = append(shutdownFailures, fmt.Errorf("关闭 HTTP 服务：%w", err))
|
||||
}
|
||||
workerCtx, workerCancel := context.WithTimeout(context.Background(), workerTimeout)
|
||||
defer workerCancel()
|
||||
if err := svc.StopWorker(workerCtx); err != nil {
|
||||
shutdownFailures = append(shutdownFailures, fmt.Errorf("等待后台任务退出：%w", err))
|
||||
}
|
||||
if err := errors.Join(shutdownFailures...); err != nil {
|
||||
return err
|
||||
}
|
||||
log.Printf("影策 backend stopped gracefully")
|
||||
return nil
|
||||
}
|
||||
|
||||
func redactCanvasSharePath(path string) string {
|
||||
@@ -138,6 +211,30 @@ func env(key string, fallback string) string {
|
||||
return value
|
||||
}
|
||||
|
||||
func envBool(key string, fallback bool) (bool, error) {
|
||||
value := strings.TrimSpace(os.Getenv(key))
|
||||
if value == "" {
|
||||
return fallback, nil
|
||||
}
|
||||
parsed, err := strconv.ParseBool(value)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("%s 必须是 true 或 false", key)
|
||||
}
|
||||
return parsed, nil
|
||||
}
|
||||
|
||||
func envDuration(key string, fallback time.Duration) (time.Duration, error) {
|
||||
value := strings.TrimSpace(os.Getenv(key))
|
||||
if value == "" {
|
||||
return fallback, nil
|
||||
}
|
||||
parsed, err := time.ParseDuration(value)
|
||||
if err != nil || parsed <= 0 {
|
||||
return 0, fmt.Errorf("%s 必须是正数时长，例如 10m", key)
|
||||
}
|
||||
return parsed, nil
|
||||
}
|
||||
|
||||
const corsAllowedHeaders = "Accept, Content-Type, Authorization, X-Requested-With, X-Canvas-Scene, X-Idempotency-Key, X-Canvas-Trace-ID, X-Canvas-Upstream-URL, X-Canvas-Upstream-Format, X-Canvas-Allow-Local-Channel, X-Canvas-Upstream-Base-URL"
|
||||
|
||||
const corsAllowedMethods = "GET, POST, PUT, PATCH, DELETE, OPTIONS"
|
||||
|
||||
@@ -3,6 +3,7 @@ package main
|
||||
import (
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
@@ -19,6 +20,19 @@ func TestAllowedOriginWildcard(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestEnvironmentParsers(t *testing.T) {
|
||||
t.Setenv("CANVAS_AUTO_MIGRATE", "false")
|
||||
value, err := envBool("CANVAS_AUTO_MIGRATE", true)
|
||||
if err != nil || value {
|
||||
t.Fatalf("envBool = %v, %v", value, err)
|
||||
}
|
||||
t.Setenv("CANVAS_SHUTDOWN_TIMEOUT", "45s")
|
||||
duration, err := envDuration("CANVAS_SHUTDOWN_TIMEOUT", time.Minute)
|
||||
if err != nil || duration != 45*time.Second {
|
||||
t.Fatalf("envDuration = %v, %v", duration, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAllowedOriginUsesForwardedHost(t *testing.T) {
|
||||
t.Setenv("CANVAS_CORS_ORIGINS", "")
|
||||
context, _ := gin.CreateTestContext(httptest.NewRecorder())
|
||||
|
||||
112
backend/cmd/server/system_status.go
Normal file
112
backend/cmd/server/system_status.go
Normal file
@@ -0,0 +1,112 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"sync/atomic"
|
||||
|
||||
"infinite-canvas/backend/internal/buildinfo"
|
||||
"infinite-canvas/backend/internal/database"
|
||||
"infinite-canvas/backend/internal/service"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
type systemStatus struct {
|
||||
db *gorm.DB
|
||||
service *service.Service
|
||||
started atomic.Bool
|
||||
draining atomic.Bool
|
||||
}
|
||||
|
||||
type systemStatusSnapshot struct {
|
||||
Status string `json:"status"`
|
||||
Ready bool `json:"ready"`
|
||||
Started bool `json:"started"`
|
||||
Draining bool `json:"draining"`
|
||||
ActiveWorkerTasks int64 `json:"activeWorkerTasks"`
|
||||
Build buildinfo.Info `json:"build"`
|
||||
Schema database.SchemaStatus `json:"schema"`
|
||||
Checks systemStatusChecks `json:"checks"`
|
||||
}
|
||||
|
||||
type systemStatusChecks struct {
|
||||
Database bool `json:"database"`
|
||||
Runtime bool `json:"runtime"`
|
||||
Schema bool `json:"schema"`
|
||||
}
|
||||
|
||||
func newSystemStatus(db *gorm.DB, svc *service.Service) *systemStatus {
|
||||
return &systemStatus{db: db, service: svc}
|
||||
}
|
||||
|
||||
func (s *systemStatus) markStarted() { s.started.Store(true) }
|
||||
|
||||
func (s *systemStatus) beginDrain() {
|
||||
s.draining.Store(true)
|
||||
if s.service != nil {
|
||||
s.service.BeginDrain()
|
||||
}
|
||||
}
|
||||
|
||||
func (s *systemStatus) snapshot(ctx context.Context) systemStatusSnapshot {
|
||||
snapshot := systemStatusSnapshot{
|
||||
Started: s.started.Load(),
|
||||
Draining: s.draining.Load(),
|
||||
Build: buildinfo.Current(),
|
||||
Schema: database.SchemaStatus{Expected: database.CurrentSchemaVersion},
|
||||
}
|
||||
if s.service != nil {
|
||||
snapshot.ActiveWorkerTasks = s.service.ActiveWorkerTasks()
|
||||
snapshot.Checks.Runtime = s.service.ValidateRuntime() == nil
|
||||
}
|
||||
if s.db != nil {
|
||||
snapshot.Checks.Database = s.db.WithContext(ctx).Exec("SELECT 1").Error == nil
|
||||
if schema, err := database.ReadSchemaStatus(s.db); err == nil {
|
||||
snapshot.Schema = schema
|
||||
snapshot.Checks.Schema = schema.Ready
|
||||
}
|
||||
}
|
||||
snapshot.Ready = snapshot.Started && !snapshot.Draining && snapshot.Checks.Database && snapshot.Checks.Runtime && snapshot.Checks.Schema
|
||||
switch {
|
||||
case snapshot.Draining:
|
||||
snapshot.Status = "draining"
|
||||
case snapshot.Ready:
|
||||
snapshot.Status = "ok"
|
||||
case !snapshot.Started:
|
||||
snapshot.Status = "starting"
|
||||
default:
|
||||
snapshot.Status = "unhealthy"
|
||||
}
|
||||
return snapshot
|
||||
}
|
||||
|
||||
func registerSystemStatusRoutes(api *gin.RouterGroup, status *systemStatus) {
|
||||
api.GET("/health/live", func(c *gin.Context) {
|
||||
c.JSON(http.StatusOK, gin.H{"code": 0, "data": gin.H{"status": "ok", "build": buildinfo.Current()}, "msg": "ok"})
|
||||
})
|
||||
startup := func(c *gin.Context) {
|
||||
snapshot := status.snapshot(c.Request.Context())
|
||||
if !snapshot.Started {
|
||||
c.JSON(http.StatusServiceUnavailable, gin.H{"code": http.StatusServiceUnavailable, "data": snapshot, "msg": "服务仍在启动"})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"code": 0, "data": snapshot, "msg": "ok"})
|
||||
}
|
||||
ready := func(c *gin.Context) {
|
||||
snapshot := status.snapshot(c.Request.Context())
|
||||
if !snapshot.Ready {
|
||||
c.JSON(http.StatusServiceUnavailable, gin.H{"code": http.StatusServiceUnavailable, "data": snapshot, "msg": "服务暂未就绪"})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusOK, gin.H{"code": 0, "data": snapshot, "msg": "ok"})
|
||||
}
|
||||
api.GET("/health/startup", startup)
|
||||
api.GET("/health/ready", ready)
|
||||
api.GET("/health", ready)
|
||||
api.GET("/system/version", func(c *gin.Context) {
|
||||
snapshot := status.snapshot(c.Request.Context())
|
||||
c.JSON(http.StatusOK, gin.H{"code": 0, "data": gin.H{"build": snapshot.Build, "schema": snapshot.Schema}, "msg": "ok"})
|
||||
})
|
||||
}
|
||||
60
backend/cmd/server/system_status_test.go
Normal file
60
backend/cmd/server/system_status_test.go
Normal file
@@ -0,0 +1,60 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"infinite-canvas/backend/internal/database"
|
||||
"infinite-canvas/backend/internal/repository"
|
||||
"infinite-canvas/backend/internal/service"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
func TestSystemReadinessTracksStartupAndDrain(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
db, err := database.Open(database.Config{Driver: "sqlite", DSN: "file:system-status?mode=memory&cache=shared"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := database.MigrateSchema(db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
svc := service.New(repository.New(db), t.TempDir())
|
||||
status := newSystemStatus(db, svc)
|
||||
router := gin.New()
|
||||
registerSystemStatusRoutes(router.Group("/api"), status)
|
||||
|
||||
assertStatus(t, router, "/api/health/live", http.StatusOK)
|
||||
assertStatus(t, router, "/api/health/ready", http.StatusServiceUnavailable)
|
||||
status.markStarted()
|
||||
assertStatus(t, router, "/api/health/ready", http.StatusOK)
|
||||
status.beginDrain()
|
||||
assertStatus(t, router, "/api/health/ready", http.StatusServiceUnavailable)
|
||||
assertStatus(t, router, "/api/health/live", http.StatusOK)
|
||||
}
|
||||
|
||||
func TestSystemReadinessRejectsMissingSchemaVersion(t *testing.T) {
|
||||
gin.SetMode(gin.TestMode)
|
||||
db, err := database.Open(database.Config{Driver: "sqlite", DSN: "file:system-status-schema?mode=memory&cache=shared"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
svc := service.New(repository.New(db), t.TempDir())
|
||||
status := newSystemStatus(db, svc)
|
||||
status.markStarted()
|
||||
router := gin.New()
|
||||
registerSystemStatusRoutes(router.Group("/api"), status)
|
||||
assertStatus(t, router, "/api/health/ready", http.StatusServiceUnavailable)
|
||||
}
|
||||
|
||||
func assertStatus(t *testing.T, handler http.Handler, path string, want int) {
|
||||
t.Helper()
|
||||
request := httptest.NewRequest(http.MethodGet, path, nil)
|
||||
recorder := httptest.NewRecorder()
|
||||
handler.ServeHTTP(recorder, request)
|
||||
if recorder.Code != want {
|
||||
t.Fatalf("GET %s status = %d, want %d; body=%s", path, recorder.Code, want, recorder.Body.String())
|
||||
}
|
||||
}
|
||||
49
backend/internal/buildinfo/buildinfo.go
Normal file
49
backend/internal/buildinfo/buildinfo.go
Normal file
@@ -0,0 +1,49 @@
|
||||
package buildinfo
|
||||
|
||||
import (
|
||||
"runtime"
|
||||
"runtime/debug"
|
||||
"strings"
|
||||
)
|
||||
|
||||
var (
|
||||
Version = "dev"
|
||||
Commit = "unknown"
|
||||
BuildTime = "unknown"
|
||||
)
|
||||
|
||||
type Info struct {
|
||||
Version string `json:"version"`
|
||||
Commit string `json:"commit"`
|
||||
BuildTime string `json:"buildTime"`
|
||||
GoVersion string `json:"goVersion"`
|
||||
}
|
||||
|
||||
func Current() Info {
|
||||
version := normalized(Version, "dev")
|
||||
commit := normalized(Commit, "unknown")
|
||||
buildTime := normalized(BuildTime, "unknown")
|
||||
if info, ok := debug.ReadBuildInfo(); ok {
|
||||
for _, setting := range info.Settings {
|
||||
switch setting.Key {
|
||||
case "vcs.revision":
|
||||
if commit == "unknown" {
|
||||
commit = normalized(setting.Value, commit)
|
||||
}
|
||||
case "vcs.time":
|
||||
if buildTime == "unknown" {
|
||||
buildTime = normalized(setting.Value, buildTime)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return Info{Version: version, Commit: commit, BuildTime: buildTime, GoVersion: runtime.Version()}
|
||||
}
|
||||
|
||||
func normalized(value string, fallback string) string {
|
||||
value = strings.TrimSpace(value)
|
||||
if value == "" {
|
||||
return fallback
|
||||
}
|
||||
return value
|
||||
}
|
||||
17
backend/internal/buildinfo/buildinfo_test.go
Normal file
17
backend/internal/buildinfo/buildinfo_test.go
Normal file
@@ -0,0 +1,17 @@
|
||||
package buildinfo
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestCurrentUsesInjectedValues(t *testing.T) {
|
||||
previousVersion, previousCommit, previousBuildTime := Version, Commit, BuildTime
|
||||
t.Cleanup(func() { Version, Commit, BuildTime = previousVersion, previousCommit, previousBuildTime })
|
||||
Version, Commit, BuildTime = " v1.2.3 ", " abc123 ", " 2026-08-30T00:00:00Z "
|
||||
|
||||
info := Current()
|
||||
if info.Version != "v1.2.3" || info.Commit != "abc123" || info.BuildTime != "2026-08-30T00:00:00Z" {
|
||||
t.Fatalf("unexpected build info: %#v", info)
|
||||
}
|
||||
if info.GoVersion == "" {
|
||||
t.Fatal("Go version should be populated")
|
||||
}
|
||||
}
|
||||
133
backend/internal/database/migrations.go
Normal file
133
backend/internal/database/migrations.go
Normal file
@@ -0,0 +1,133 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
const CurrentSchemaVersion int64 = 1
|
||||
|
||||
const baselineSchemaChecksum = "sha256:open-ai-canvas-schema-v1-20260830"
|
||||
|
||||
const postgresSchemaMigrationLockID int64 = 73123910420260830
|
||||
|
||||
type SchemaStatus struct {
|
||||
Current int64 `json:"current"`
|
||||
Expected int64 `json:"expected"`
|
||||
Ready bool `json:"ready"`
|
||||
}
|
||||
|
||||
type schemaMigration struct {
|
||||
Version int64 `gorm:"primaryKey"`
|
||||
Name string `gorm:"size:160;not null"`
|
||||
Checksum string `gorm:"size:96;not null"`
|
||||
AppliedAt time.Time `gorm:"not null"`
|
||||
}
|
||||
|
||||
func (schemaMigration) TableName() string { return "schema_migrations" }
|
||||
|
||||
type migration struct {
|
||||
version int64
|
||||
name string
|
||||
checksum string
|
||||
apply func(*gorm.DB) error
|
||||
}
|
||||
|
||||
var schemaMigrations = []migration{
|
||||
{version: 1, name: "baseline_gorm_schema", checksum: baselineSchemaChecksum, apply: migrateSchemaV1},
|
||||
}
|
||||
|
||||
func MigrateSchema(db *gorm.DB) error {
|
||||
return db.Transaction(func(tx *gorm.DB) error {
|
||||
if tx.Dialector.Name() == "postgres" {
|
||||
if err := tx.Exec("SELECT pg_advisory_xact_lock(?)", postgresSchemaMigrationLockID).Error; err != nil {
|
||||
return fmt.Errorf("获取数据库迁移锁：%w", err)
|
||||
}
|
||||
}
|
||||
if err := tx.AutoMigrate(&schemaMigration{}); err != nil {
|
||||
return fmt.Errorf("初始化数据库迁移记录：%w", err)
|
||||
}
|
||||
for _, item := range schemaMigrations {
|
||||
var applied schemaMigration
|
||||
err := tx.First(&applied, "version = ?", item.version).Error
|
||||
if err == nil {
|
||||
if err := validateMigrationRecord(applied, item); err != nil {
|
||||
return err
|
||||
}
|
||||
continue
|
||||
}
|
||||
if !errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return fmt.Errorf("读取数据库迁移 %d：%w", item.version, err)
|
||||
}
|
||||
if err := item.apply(tx); err != nil {
|
||||
return fmt.Errorf("执行数据库迁移 %d（%s）：%w", item.version, item.name, err)
|
||||
}
|
||||
record := schemaMigration{Version: item.version, Name: item.name, Checksum: item.checksum, AppliedAt: time.Now().UTC()}
|
||||
if err := tx.Create(&record).Error; err != nil {
|
||||
return fmt.Errorf("记录数据库迁移 %d：%w", item.version, err)
|
||||
}
|
||||
}
|
||||
return RequireSchemaVersion(tx)
|
||||
})
|
||||
}
|
||||
|
||||
func ReadSchemaStatus(db *gorm.DB) (SchemaStatus, error) {
|
||||
status := SchemaStatus{Expected: CurrentSchemaVersion}
|
||||
if !db.Migrator().HasTable(&schemaMigration{}) {
|
||||
return status, nil
|
||||
}
|
||||
if err := db.Model(&schemaMigration{}).Select("COALESCE(MAX(version), 0)").Scan(&status.Current).Error; err != nil {
|
||||
return status, fmt.Errorf("读取数据库结构版本：%w", err)
|
||||
}
|
||||
if status.Current != status.Expected {
|
||||
return status, nil
|
||||
}
|
||||
if err := validateMigrationRecords(db); err != nil {
|
||||
return status, err
|
||||
}
|
||||
status.Ready = true
|
||||
return status, nil
|
||||
}
|
||||
|
||||
func validateMigrationRecords(db *gorm.DB) error {
|
||||
for _, item := range schemaMigrations {
|
||||
var applied schemaMigration
|
||||
if err := db.First(&applied, "version = ?", item.version).Error; err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
return fmt.Errorf("数据库缺少迁移记录 %d（%s）", item.version, item.name)
|
||||
}
|
||||
return fmt.Errorf("读取数据库迁移 %d：%w", item.version, err)
|
||||
}
|
||||
if err := validateMigrationRecord(applied, item); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func validateMigrationRecord(applied schemaMigration, expected migration) error {
|
||||
if applied.Name != expected.name {
|
||||
return fmt.Errorf("数据库迁移 %d 名称不一致：记录为 %s，程序期望 %s", expected.version, applied.Name, expected.name)
|
||||
}
|
||||
if applied.Checksum != expected.checksum {
|
||||
return fmt.Errorf("数据库迁移 %d 校验和不一致：记录为 %s，程序期望 %s", expected.version, applied.Checksum, expected.checksum)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func RequireSchemaVersion(db *gorm.DB) error {
|
||||
status, err := ReadSchemaStatus(db)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if status.Current < status.Expected {
|
||||
return fmt.Errorf("数据库结构版本过旧：当前 %d，程序要求 %d，请先执行 migrate-schema up", status.Current, status.Expected)
|
||||
}
|
||||
if status.Current > status.Expected {
|
||||
return fmt.Errorf("数据库结构版本 %d 高于程序支持的 %d，拒绝使用旧程序连接新数据库", status.Current, status.Expected)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
96
backend/internal/database/migrations_test.go
Normal file
96
backend/internal/database/migrations_test.go
Normal file
@@ -0,0 +1,96 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func TestMigrateSchemaRecordsAndValidatesVersion(t *testing.T) {
|
||||
db, err := Open(Config{Driver: "sqlite", DSN: "file:migration-version?mode=memory&cache=shared"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := MigrateSchema(db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
status, err := ReadSchemaStatus(db)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !status.Ready || status.Current != CurrentSchemaVersion {
|
||||
t.Fatalf("unexpected schema status: %#v", status)
|
||||
}
|
||||
if err := MigrateSchema(db); err != nil {
|
||||
t.Fatalf("migration should be idempotent: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMigrateSchemaRejectsChecksumMismatch(t *testing.T) {
|
||||
db, err := Open(Config{Driver: "sqlite", DSN: "file:migration-checksum?mode=memory&cache=shared"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := MigrateSchema(db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := db.Model(&schemaMigration{}).Where("version = ?", CurrentSchemaVersion).Update("checksum", "changed").Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := MigrateSchema(db); err == nil || !strings.Contains(err.Error(), "校验和不一致") {
|
||||
t.Fatalf("expected checksum mismatch, got %v", err)
|
||||
}
|
||||
if err := RequireSchemaVersion(db); err == nil || !strings.Contains(err.Error(), "校验和不一致") {
|
||||
t.Fatalf("schema verification must reject checksum mismatch, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMigrateSchemaRollsBackFailedMigration(t *testing.T) {
|
||||
db, err := Open(Config{Driver: "sqlite", DSN: "file:migration-rollback?mode=memory&cache=shared"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := MigrateSchema(db); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
original := schemaMigrations
|
||||
schemaMigrations = append(append([]migration(nil), original...), migration{
|
||||
version: CurrentSchemaVersion + 1,
|
||||
name: "rollback_probe",
|
||||
checksum: "sha256:rollback-probe",
|
||||
apply: func(tx *gorm.DB) error {
|
||||
if err := tx.Exec("CREATE TABLE migration_rollback_probe (id INTEGER PRIMARY KEY)").Error; err != nil {
|
||||
return err
|
||||
}
|
||||
return errors.New("forced migration failure")
|
||||
},
|
||||
})
|
||||
t.Cleanup(func() { schemaMigrations = original })
|
||||
|
||||
if err := MigrateSchema(db); err == nil || !strings.Contains(err.Error(), "forced migration failure") {
|
||||
t.Fatalf("expected forced migration failure, got %v", err)
|
||||
}
|
||||
if db.Migrator().HasTable("migration_rollback_probe") {
|
||||
t.Fatal("failed migration left a partial table behind")
|
||||
}
|
||||
var count int64
|
||||
if err := db.Model(&schemaMigration{}).Where("version = ?", CurrentSchemaVersion+1).Count(&count).Error; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if count != 0 {
|
||||
t.Fatalf("failed migration was recorded: %d", count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequireSchemaVersionRejectsUninitializedDatabase(t *testing.T) {
|
||||
db, err := Open(Config{Driver: "sqlite", DSN: "file:migration-uninitialized?mode=memory&cache=shared"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := RequireSchemaVersion(db); err == nil || !strings.Contains(err.Error(), "请先执行 migrate-schema up") {
|
||||
t.Fatalf("expected missing migration error, got %v", err)
|
||||
}
|
||||
}
|
||||
@@ -91,7 +91,7 @@ func Models() []any {
|
||||
}
|
||||
}
|
||||
|
||||
func MigrateSchema(db *gorm.DB) error {
|
||||
func migrateSchemaV1(db *gorm.DB) error {
|
||||
// 旧表只保存 Updream 目录状态，与本地技能主键没有可迁移关系；首次升级时按产品要求清空重建。
|
||||
if db.Migrator().HasColumn(&model.UserSkillState{}, "skill_dir") && !db.Migrator().HasColumn(&model.UserSkillState{}, "skill_id") {
|
||||
if err := db.Migrator().DropTable(&model.UserSkillState{}); err != nil {
|
||||
|
||||
80
backend/internal/handler/admin_update.go
Normal file
80
backend/internal/handler/admin_update.go
Normal file
@@ -0,0 +1,80 @@
|
||||
package handler
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"infinite-canvas/backend/internal/service"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
func RegisterAdminUpdateRoutes(r *gin.RouterGroup, svc *service.Service) {
|
||||
r.GET("/admin/system-update", func(c *gin.Context) {
|
||||
actor, err := currentUser(c, svc)
|
||||
if err != nil {
|
||||
failService(c, err)
|
||||
return
|
||||
}
|
||||
status, err := svc.AdminUpdateStatus(c.Request.Context(), actor)
|
||||
if err != nil {
|
||||
failService(c, err)
|
||||
return
|
||||
}
|
||||
ok(c, status)
|
||||
})
|
||||
r.POST("/admin/system-update/check", func(c *gin.Context) {
|
||||
actor, err := currentUser(c, svc)
|
||||
if err != nil {
|
||||
failService(c, err)
|
||||
return
|
||||
}
|
||||
status, err := svc.AdminCheckUpdate(c.Request.Context(), actor)
|
||||
if err != nil {
|
||||
failService(c, err)
|
||||
return
|
||||
}
|
||||
ok(c, status)
|
||||
})
|
||||
r.POST("/admin/system-update/start", func(c *gin.Context) {
|
||||
actor, err := currentUser(c, svc)
|
||||
if err != nil {
|
||||
failService(c, err)
|
||||
return
|
||||
}
|
||||
c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, 16<<10)
|
||||
var request struct {
|
||||
TargetVersion string `json:"targetVersion" binding:"required"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&request); err != nil {
|
||||
fail(c, http.StatusBadRequest, err)
|
||||
return
|
||||
}
|
||||
status, err := svc.AdminStartUpdate(c.Request.Context(), actor, request.TargetVersion)
|
||||
if err != nil {
|
||||
failService(c, err)
|
||||
return
|
||||
}
|
||||
ok(c, status)
|
||||
})
|
||||
r.POST("/admin/system-update/rollback", func(c *gin.Context) {
|
||||
actor, err := currentUser(c, svc)
|
||||
if err != nil {
|
||||
failService(c, err)
|
||||
return
|
||||
}
|
||||
c.Request.Body = http.MaxBytesReader(c.Writer, c.Request.Body, 16<<10)
|
||||
var request struct {
|
||||
Reason string `json:"reason" binding:"required"`
|
||||
}
|
||||
if err := c.ShouldBindJSON(&request); err != nil {
|
||||
fail(c, http.StatusBadRequest, err)
|
||||
return
|
||||
}
|
||||
status, err := svc.AdminRollbackUpdate(c.Request.Context(), actor, request.Reason)
|
||||
if err != nil {
|
||||
failService(c, err)
|
||||
return
|
||||
}
|
||||
ok(c, status)
|
||||
})
|
||||
}
|
||||
550
backend/internal/hostupdate/manager.go
Normal file
550
backend/internal/hostupdate/manager.go
Normal file
@@ -0,0 +1,550 @@
|
||||
package hostupdate
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
const maxReleaseResponseBytes = 4 << 20
|
||||
|
||||
type Config struct {
|
||||
Repository string
|
||||
InstallDir string
|
||||
ComposeFile string
|
||||
EnvFile string
|
||||
StateDir string
|
||||
BackupDir string
|
||||
HealthURL string
|
||||
GitHubToken string
|
||||
StableWindow time.Duration
|
||||
StepTimeout time.Duration
|
||||
BinaryPath string
|
||||
ServiceName string
|
||||
SelfUpdate bool
|
||||
}
|
||||
|
||||
type commandRunner interface {
|
||||
Run(context.Context, string, []string, []string, io.Writer, io.Writer) error
|
||||
}
|
||||
|
||||
type execRunner struct{ dir string }
|
||||
|
||||
func (r execRunner) Run(ctx context.Context, name string, args, environment []string, stdout, stderr io.Writer) error {
|
||||
command := exec.CommandContext(ctx, name, args...)
|
||||
command.Dir = r.dir
|
||||
command.Env = append(os.Environ(), environment...)
|
||||
command.Stdout = stdout
|
||||
command.Stderr = stderr
|
||||
return command.Run()
|
||||
}
|
||||
|
||||
type persistedState struct {
|
||||
LatestRelease *Release `json:"latestRelease,omitempty"`
|
||||
LastBackup *Backup `json:"lastBackup,omitempty"`
|
||||
RollbackVersion string `json:"rollbackVersion,omitempty"`
|
||||
Operation Operation `json:"operation"`
|
||||
}
|
||||
|
||||
type Manager struct {
|
||||
mu sync.Mutex
|
||||
config Config
|
||||
runner commandRunner
|
||||
httpClient *http.Client
|
||||
state persistedState
|
||||
}
|
||||
|
||||
func NewManager(config Config) (*Manager, error) {
|
||||
config.Repository = strings.TrimSpace(config.Repository)
|
||||
if config.Repository == "" {
|
||||
config.Repository = "ddcat-ai/open-ai-canvas"
|
||||
}
|
||||
if config.InstallDir == "" {
|
||||
config.InstallDir = "/opt/open-ai-canvas"
|
||||
}
|
||||
if config.ComposeFile == "" {
|
||||
config.ComposeFile = "docker-compose.deploy.yml"
|
||||
}
|
||||
if config.EnvFile == "" {
|
||||
config.EnvFile = ".env"
|
||||
}
|
||||
if config.StateDir == "" {
|
||||
config.StateDir = "/var/lib/open-ai-canvas-updater"
|
||||
}
|
||||
if config.BackupDir == "" {
|
||||
config.BackupDir = filepath.Join(config.InstallDir, "backups")
|
||||
}
|
||||
if config.StableWindow <= 0 {
|
||||
config.StableWindow = 30 * time.Second
|
||||
}
|
||||
if config.StepTimeout <= 0 {
|
||||
config.StepTimeout = 20 * time.Minute
|
||||
}
|
||||
if err := os.MkdirAll(config.StateDir, 0o700); err != nil {
|
||||
return nil, fmt.Errorf("创建更新器状态目录：%w", err)
|
||||
}
|
||||
if err := os.MkdirAll(config.BackupDir, 0o700); err != nil {
|
||||
return nil, fmt.Errorf("创建备份目录：%w", err)
|
||||
}
|
||||
manager := &Manager{
|
||||
config: config,
|
||||
runner: execRunner{dir: config.InstallDir},
|
||||
httpClient: &http.Client{Timeout: 30 * time.Second},
|
||||
state: persistedState{Operation: Operation{Phase: PhaseIdle, Logs: []LogEntry{}}},
|
||||
}
|
||||
if err := manager.loadState(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if manager.state.Operation.Phase.Active() {
|
||||
manager.state.Operation.Phase = PhaseManualIntervention
|
||||
manager.state.Operation.Error = "更新器进程在操作期间退出，请检查容器与数据库状态后手动处理"
|
||||
manager.appendLogLocked(PhaseManualIntervention, manager.state.Operation.Error)
|
||||
_ = manager.saveStateLocked()
|
||||
}
|
||||
return manager, nil
|
||||
}
|
||||
|
||||
func (m *Manager) Snapshot() Status {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
return m.snapshotLocked()
|
||||
}
|
||||
|
||||
func (m *Manager) Check(ctx context.Context) (Status, error) {
|
||||
m.mu.Lock()
|
||||
if m.state.Operation.Phase.Active() {
|
||||
status := m.snapshotLocked()
|
||||
m.mu.Unlock()
|
||||
return status, errors.New("更新操作正在进行，暂时不能检查新版本")
|
||||
}
|
||||
m.state.Operation.Phase = PhaseChecking
|
||||
m.state.Operation.Error = ""
|
||||
m.appendLogLocked(PhaseChecking, "正在读取 GitHub Release")
|
||||
_ = m.saveStateLocked()
|
||||
m.mu.Unlock()
|
||||
|
||||
release, err := m.latestRelease(ctx)
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if err != nil {
|
||||
m.state.Operation.Phase = PhaseFailed
|
||||
m.state.Operation.Error = err.Error()
|
||||
m.appendLogLocked(PhaseFailed, "检查更新失败")
|
||||
_ = m.saveStateLocked()
|
||||
return m.snapshotLocked(), err
|
||||
}
|
||||
m.state.LatestRelease = release
|
||||
current, currentErr := m.currentVersion()
|
||||
if currentErr != nil {
|
||||
m.state.Operation.Phase = PhaseFailed
|
||||
m.state.Operation.Error = currentErr.Error()
|
||||
_ = m.saveStateLocked()
|
||||
return m.snapshotLocked(), currentErr
|
||||
}
|
||||
if CompareVersions(current, release.Version) < 0 {
|
||||
m.state.Operation.Phase = PhaseReady
|
||||
m.appendLogLocked(PhaseReady, fmt.Sprintf("发现新版本 %s", release.Version))
|
||||
} else {
|
||||
m.state.Operation.Phase = PhaseNoUpdate
|
||||
m.appendLogLocked(PhaseNoUpdate, "当前已是最新版本")
|
||||
}
|
||||
_ = m.saveStateLocked()
|
||||
return m.snapshotLocked(), nil
|
||||
}
|
||||
|
||||
func (m *Manager) StartUpdate(targetVersion string) (Status, error) {
|
||||
targetVersion = strings.TrimSpace(targetVersion)
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if m.state.Operation.Phase.Active() {
|
||||
return m.snapshotLocked(), errors.New("已有更新操作正在进行")
|
||||
}
|
||||
if m.state.LatestRelease == nil || m.state.LatestRelease.Version != targetVersion {
|
||||
return m.snapshotLocked(), errors.New("目标版本与最近一次检查结果不一致，请重新检查更新")
|
||||
}
|
||||
current, err := m.currentVersion()
|
||||
if err != nil {
|
||||
return m.snapshotLocked(), err
|
||||
}
|
||||
if CompareVersions(current, targetVersion) >= 0 {
|
||||
return m.snapshotLocked(), errors.New("目标版本必须高于当前版本")
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
m.state.Operation = Operation{
|
||||
ID: randomID(),
|
||||
Phase: PhasePreflight,
|
||||
FromVersion: current,
|
||||
TargetVersion: targetVersion,
|
||||
StartedAt: &now,
|
||||
Logs: []LogEntry{{At: now, Phase: PhasePreflight, Message: "开始更新前检查"}},
|
||||
}
|
||||
_ = m.saveStateLocked()
|
||||
go m.runUpdate(current, targetVersion)
|
||||
return m.snapshotLocked(), nil
|
||||
}
|
||||
|
||||
func (m *Manager) StartRollback(reason string) (Status, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if m.state.Operation.Phase.Active() {
|
||||
return m.snapshotLocked(), errors.New("已有更新操作正在进行")
|
||||
}
|
||||
if m.state.RollbackVersion == "" || m.state.LastBackup == nil {
|
||||
return m.snapshotLocked(), errors.New("没有可用的回退版本或已校验备份")
|
||||
}
|
||||
current, err := m.currentVersion()
|
||||
if err != nil {
|
||||
return m.snapshotLocked(), err
|
||||
}
|
||||
now := time.Now().UTC()
|
||||
target := m.state.RollbackVersion
|
||||
m.state.Operation = Operation{
|
||||
ID: randomID(),
|
||||
Phase: PhaseRollingBack,
|
||||
FromVersion: current,
|
||||
TargetVersion: target,
|
||||
StartedAt: &now,
|
||||
Logs: []LogEntry{{At: now, Phase: PhaseRollingBack, Message: "管理员发起人工回退：" + strings.TrimSpace(reason)}},
|
||||
}
|
||||
_ = m.saveStateLocked()
|
||||
backup := *m.state.LastBackup
|
||||
go m.runRollback(target, backup, false)
|
||||
return m.snapshotLocked(), nil
|
||||
}
|
||||
|
||||
func (m *Manager) runUpdate(fromVersion, targetVersion string) {
|
||||
nextCompose, err := m.prepareTargetCompose(targetVersion)
|
||||
if err != nil {
|
||||
m.failWithoutRollback(PhaseFailed, err)
|
||||
return
|
||||
}
|
||||
defer os.Remove(nextCompose)
|
||||
if err := m.preflight(nextCompose, targetVersion); err != nil {
|
||||
m.failWithoutRollback(PhaseFailed, err)
|
||||
return
|
||||
}
|
||||
updaterBinary, err := m.prepareUpdaterBinary(targetVersion)
|
||||
if err != nil {
|
||||
m.failWithoutRollback(PhaseFailed, err)
|
||||
return
|
||||
}
|
||||
if updaterBinary != "" {
|
||||
defer os.Remove(updaterBinary)
|
||||
}
|
||||
|
||||
m.setPhase(PhaseBackingUp, "创建 PostgreSQL 与数据目录 ZIP 备份")
|
||||
backup, err := m.createBackup(fromVersion)
|
||||
if err != nil {
|
||||
m.failWithoutRollback(PhaseFailed, err)
|
||||
return
|
||||
}
|
||||
m.mu.Lock()
|
||||
m.state.LastBackup = &backup
|
||||
m.state.RollbackVersion = fromVersion
|
||||
_ = m.saveStateLocked()
|
||||
m.mu.Unlock()
|
||||
if err := replaceFile(m.composePath(), m.previousComposePath(), 0o600); err != nil {
|
||||
m.failWithoutRollback(PhaseFailed, fmt.Errorf("保存旧 Compose 配置：%w", err))
|
||||
return
|
||||
}
|
||||
|
||||
m.setPhase(PhasePulling, "拉取目标版本镜像")
|
||||
if err := m.compose(nextCompose, targetVersion, m.config.StepTimeout, nil, "pull", "backend", "web"); err != nil {
|
||||
m.failWithoutRollback(PhaseFailed, err)
|
||||
return
|
||||
}
|
||||
if err := m.verifyImages(targetVersion); err != nil {
|
||||
m.failWithoutRollback(PhaseFailed, err)
|
||||
return
|
||||
}
|
||||
|
||||
m.setPhase(PhaseDraining, "停止 Web 与 Backend，等待后台任务优雅退出")
|
||||
if err := m.compose(m.composePath(), fromVersion, m.config.StepTimeout, nil, "stop", "web", "backend"); err != nil {
|
||||
m.failWithRollback(err, fromVersion, backup)
|
||||
return
|
||||
}
|
||||
|
||||
m.setPhase(PhaseMigrating, "执行目标版本数据库迁移")
|
||||
if err := m.compose(nextCompose, targetVersion, m.config.StepTimeout, nil, "run", "--rm", "migrate"); err != nil {
|
||||
m.failWithRollback(err, fromVersion, backup)
|
||||
return
|
||||
}
|
||||
|
||||
m.setPhase(PhaseSwitching, "切换 Compose 配置与镜像版本")
|
||||
if err := setEnvValue(m.envPath(), "CANVAS_IMAGE_TAG", strings.TrimPrefix(targetVersion, "v")); err != nil {
|
||||
m.failWithRollback(err, fromVersion, backup)
|
||||
return
|
||||
}
|
||||
if err := replaceFile(nextCompose, m.composePath(), 0o600); err != nil {
|
||||
_ = setEnvValue(m.envPath(), "CANVAS_IMAGE_TAG", strings.TrimPrefix(fromVersion, "v"))
|
||||
m.failWithRollback(err, fromVersion, backup)
|
||||
return
|
||||
}
|
||||
if err := m.compose(m.composePath(), targetVersion, m.config.StepTimeout, nil, "up", "-d", "--remove-orphans", "--wait", "--wait-timeout", "600"); err != nil {
|
||||
m.failWithRollback(err, fromVersion, backup)
|
||||
return
|
||||
}
|
||||
|
||||
m.setPhase(PhaseVerifying, "验证健康状态、运行版本和稳定窗口")
|
||||
if err := m.verifyHealthy(targetVersion); err != nil {
|
||||
m.failWithRollback(err, fromVersion, backup)
|
||||
return
|
||||
}
|
||||
if updaterBinary != "" {
|
||||
if err := replaceFile(updaterBinary, m.config.BinaryPath, 0o755); err != nil {
|
||||
m.mu.Lock()
|
||||
now := time.Now().UTC()
|
||||
m.state.Operation.Phase = PhaseManualIntervention
|
||||
m.state.Operation.FinishedAt = &now
|
||||
m.state.Operation.Error = "应用已更新，但 Host Updater 自更新失败：" + safeOperationError(err)
|
||||
m.appendLogLocked(PhaseManualIntervention, "应用版本已切换，更新器二进制需要人工更新")
|
||||
_ = m.saveStateLocked()
|
||||
m.mu.Unlock()
|
||||
return
|
||||
}
|
||||
}
|
||||
m.mu.Lock()
|
||||
now := time.Now().UTC()
|
||||
m.state.Operation.Phase = PhaseSucceeded
|
||||
m.state.Operation.FinishedAt = &now
|
||||
m.state.Operation.Error = ""
|
||||
m.appendLogLocked(PhaseSucceeded, fmt.Sprintf("已成功更新到 %s", targetVersion))
|
||||
if updaterBinary != "" {
|
||||
m.appendLogLocked(PhaseSucceeded, "Host Updater 二进制已同步到目标 Release")
|
||||
}
|
||||
_ = m.saveStateLocked()
|
||||
m.mu.Unlock()
|
||||
if updaterBinary != "" && m.config.ServiceName != "" {
|
||||
go m.restartSelf()
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) failWithoutRollback(phase Phase, err error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
now := time.Now().UTC()
|
||||
m.state.Operation.Phase = phase
|
||||
m.state.Operation.Error = safeOperationError(err)
|
||||
m.state.Operation.FinishedAt = &now
|
||||
m.appendLogLocked(phase, "更新已中止，尚未修改运行版本")
|
||||
_ = m.saveStateLocked()
|
||||
}
|
||||
|
||||
func (m *Manager) failWithRollback(cause error, targetVersion string, backup Backup) {
|
||||
m.mu.Lock()
|
||||
m.state.Operation.Error = safeOperationError(cause)
|
||||
m.state.Operation.AutomaticRollback = true
|
||||
m.state.Operation.Phase = PhaseRollingBack
|
||||
m.appendLogLocked(PhaseRollingBack, "更新失败，正在自动恢复旧版本和数据库备份")
|
||||
_ = m.saveStateLocked()
|
||||
m.mu.Unlock()
|
||||
m.runRollback(targetVersion, backup, true)
|
||||
}
|
||||
|
||||
func (m *Manager) runRollback(targetVersion string, backup Backup, automatic bool) {
|
||||
var failures []error
|
||||
if err := m.compose(m.composePath(), targetVersion, 5*time.Minute, nil, "stop", "web", "backend"); err != nil {
|
||||
failures = append(failures, err)
|
||||
}
|
||||
if _, err := os.Stat(m.previousComposePath()); err == nil {
|
||||
if err := replaceFile(m.previousComposePath(), m.composePath(), 0o600); err != nil {
|
||||
failures = append(failures, fmt.Errorf("恢复旧 Compose 配置：%w", err))
|
||||
}
|
||||
}
|
||||
if err := m.restoreDatabase(backup); err != nil {
|
||||
failures = append(failures, err)
|
||||
}
|
||||
if err := setEnvValue(m.envPath(), "CANVAS_IMAGE_TAG", strings.TrimPrefix(targetVersion, "v")); err != nil {
|
||||
failures = append(failures, err)
|
||||
}
|
||||
started := true
|
||||
if err := m.compose(m.composePath(), targetVersion, m.config.StepTimeout, nil, "up", "-d", "--remove-orphans", "--wait", "--wait-timeout", "600"); err != nil {
|
||||
started = false
|
||||
failures = append(failures, err)
|
||||
}
|
||||
if started {
|
||||
if err := m.verifyHealthy(targetVersion); err != nil {
|
||||
failures = append(failures, err)
|
||||
}
|
||||
}
|
||||
rollbackErr := errors.Join(failures...)
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
now := time.Now().UTC()
|
||||
m.state.Operation.FinishedAt = &now
|
||||
m.state.Operation.AutomaticRollback = automatic
|
||||
if rollbackErr != nil {
|
||||
m.state.Operation.Phase = PhaseManualIntervention
|
||||
m.state.Operation.RollbackError = safeOperationError(rollbackErr)
|
||||
m.appendLogLocked(PhaseManualIntervention, "自动回退未能完成，需要人工介入")
|
||||
} else {
|
||||
m.state.Operation.Phase = PhaseRolledBack
|
||||
m.state.Operation.RollbackError = ""
|
||||
m.appendLogLocked(PhaseRolledBack, fmt.Sprintf("已恢复到 %s", targetVersion))
|
||||
}
|
||||
_ = m.saveStateLocked()
|
||||
}
|
||||
|
||||
func (m *Manager) snapshotLocked() Status {
|
||||
current, err := m.currentVersion()
|
||||
checks := m.checks(current, err)
|
||||
status := Status{
|
||||
Supported: true,
|
||||
Connected: true,
|
||||
Repository: m.config.Repository,
|
||||
Deployment: "docker-compose-host-updater",
|
||||
CurrentVersion: current,
|
||||
LatestRelease: cloneRelease(m.state.LatestRelease),
|
||||
Checks: checks,
|
||||
LastBackup: cloneBackup(m.state.LastBackup),
|
||||
RollbackVersion: m.state.RollbackVersion,
|
||||
Operation: cloneOperation(m.state.Operation),
|
||||
}
|
||||
if status.LatestRelease != nil && err == nil {
|
||||
status.UpdateAvailable = CompareVersions(current, status.LatestRelease.Version) < 0
|
||||
}
|
||||
return status
|
||||
}
|
||||
|
||||
func (m *Manager) checks(current string, currentErr error) []Check {
|
||||
items := []Check{
|
||||
{Key: "updater", Label: "Host Updater", Status: "passed", Detail: runtime.GOOS + "/" + runtime.GOARCH, Blocking: true},
|
||||
{Key: "version", Label: "当前版本", Status: "passed", Detail: current, Blocking: true},
|
||||
{Key: "backup", Label: "数据库备份 ZIP", Status: "pending", Detail: "开始更新后自动创建并校验", Blocking: true},
|
||||
{Key: "images", Label: "目标镜像", Status: "pending", Detail: "拉取后校验仓库摘要", Blocking: true},
|
||||
{Key: "migration", Label: "数据库迁移", Status: "pending", Detail: "在旧服务完全停止后执行", Blocking: true},
|
||||
}
|
||||
if currentErr != nil {
|
||||
items[1].Status = "failed"
|
||||
items[1].Detail = currentErr.Error()
|
||||
}
|
||||
if m.state.LastBackup != nil {
|
||||
items[2].Status = "passed"
|
||||
items[2].Detail = m.state.LastBackup.ID + " · " + m.state.LastBackup.Checksum
|
||||
}
|
||||
return items
|
||||
}
|
||||
|
||||
func (m *Manager) setPhase(phase Phase, message string) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.state.Operation.Phase = phase
|
||||
m.appendLogLocked(phase, message)
|
||||
_ = m.saveStateLocked()
|
||||
}
|
||||
|
||||
func (m *Manager) appendLogLocked(phase Phase, message string) {
|
||||
m.state.Operation.Logs = append(m.state.Operation.Logs, LogEntry{At: time.Now().UTC(), Phase: phase, Message: message})
|
||||
if len(m.state.Operation.Logs) > 200 {
|
||||
m.state.Operation.Logs = append([]LogEntry(nil), m.state.Operation.Logs[len(m.state.Operation.Logs)-200:]...)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) loadState() error {
|
||||
data, err := os.ReadFile(m.statePath())
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("读取更新器状态：%w", err)
|
||||
}
|
||||
if err := json.Unmarshal(data, &m.state); err != nil {
|
||||
return fmt.Errorf("解析更新器状态：%w", err)
|
||||
}
|
||||
if m.state.Operation.Phase == "" {
|
||||
m.state.Operation.Phase = PhaseIdle
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) saveStateLocked() error {
|
||||
data, err := json.MarshalIndent(m.state, "", " ")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
temporary, err := os.CreateTemp(m.config.StateDir, ".state-*.json")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
temporaryName := temporary.Name()
|
||||
defer os.Remove(temporaryName)
|
||||
if err := temporary.Chmod(0o600); err != nil {
|
||||
temporary.Close()
|
||||
return err
|
||||
}
|
||||
if _, err := temporary.Write(data); err != nil {
|
||||
temporary.Close()
|
||||
return err
|
||||
}
|
||||
if err := temporary.Sync(); err != nil {
|
||||
temporary.Close()
|
||||
return err
|
||||
}
|
||||
if err := temporary.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.Rename(temporaryName, m.statePath())
|
||||
}
|
||||
|
||||
func (m *Manager) statePath() string { return filepath.Join(m.config.StateDir, "state.json") }
|
||||
func (m *Manager) composePath() string {
|
||||
return filepath.Join(m.config.InstallDir, m.config.ComposeFile)
|
||||
}
|
||||
func (m *Manager) envPath() string { return filepath.Join(m.config.InstallDir, m.config.EnvFile) }
|
||||
func (m *Manager) previousComposePath() string {
|
||||
return filepath.Join(m.config.StateDir, "previous-compose.yml")
|
||||
}
|
||||
|
||||
func randomID() string {
|
||||
buffer := make([]byte, 12)
|
||||
if _, err := rand.Read(buffer); err != nil {
|
||||
return strconv.FormatInt(time.Now().UnixNano(), 36)
|
||||
}
|
||||
return hex.EncodeToString(buffer)
|
||||
}
|
||||
|
||||
func safeOperationError(err error) string {
|
||||
if err == nil {
|
||||
return ""
|
||||
}
|
||||
message := strings.TrimSpace(err.Error())
|
||||
if len(message) > 1200 {
|
||||
message = message[:1200]
|
||||
}
|
||||
return message
|
||||
}
|
||||
|
||||
func cloneRelease(value *Release) *Release {
|
||||
if value == nil {
|
||||
return nil
|
||||
}
|
||||
cloned := *value
|
||||
return &cloned
|
||||
}
|
||||
|
||||
func cloneBackup(value *Backup) *Backup {
|
||||
if value == nil {
|
||||
return nil
|
||||
}
|
||||
cloned := *value
|
||||
cloned.Path = ""
|
||||
return &cloned
|
||||
}
|
||||
|
||||
func cloneOperation(value Operation) Operation {
|
||||
value.Logs = append([]LogEntry(nil), value.Logs...)
|
||||
return value
|
||||
}
|
||||
612
backend/internal/hostupdate/manager_ops.go
Normal file
612
backend/internal/hostupdate/manager_ops.go
Normal file
@@ -0,0 +1,612 @@
|
||||
package hostupdate
|
||||
|
||||
import (
|
||||
"archive/zip"
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"runtime"
|
||||
"sort"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
)
|
||||
|
||||
type githubRelease struct {
|
||||
TagName string `json:"tag_name"`
|
||||
Name string `json:"name"`
|
||||
Body string `json:"body"`
|
||||
HTMLURL string `json:"html_url"`
|
||||
PublishedAt time.Time `json:"published_at"`
|
||||
Draft bool `json:"draft"`
|
||||
Prerelease bool `json:"prerelease"`
|
||||
}
|
||||
|
||||
func (m *Manager) latestRelease(ctx context.Context) (*Release, error) {
|
||||
url := "https://api.github.com/repos/" + m.config.Repository + "/releases?per_page=30"
|
||||
request, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
request.Header.Set("Accept", "application/vnd.github+json")
|
||||
request.Header.Set("User-Agent", "open-ai-canvas-host-updater")
|
||||
request.Header.Set("X-GitHub-Api-Version", "2022-11-28")
|
||||
if m.config.GitHubToken != "" {
|
||||
request.Header.Set("Authorization", "Bearer "+m.config.GitHubToken)
|
||||
}
|
||||
response, err := m.httpClient.Do(request)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("请求 GitHub Release：%w", err)
|
||||
}
|
||||
defer response.Body.Close()
|
||||
if response.StatusCode != http.StatusOK {
|
||||
_, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 64<<10))
|
||||
return nil, fmt.Errorf("GitHub Release 返回 HTTP %d", response.StatusCode)
|
||||
}
|
||||
var releases []githubRelease
|
||||
decoder := json.NewDecoder(io.LimitReader(response.Body, maxReleaseResponseBytes))
|
||||
if err := decoder.Decode(&releases); err != nil {
|
||||
return nil, fmt.Errorf("解析 GitHub Release：%w", err)
|
||||
}
|
||||
filtered := make([]githubRelease, 0, len(releases))
|
||||
for _, release := range releases {
|
||||
if !release.Draft && strings.HasPrefix(release.TagName, "v") {
|
||||
filtered = append(filtered, release)
|
||||
}
|
||||
}
|
||||
if len(filtered) == 0 {
|
||||
return nil, errors.New("GitHub 尚未发布可用 Release")
|
||||
}
|
||||
sort.SliceStable(filtered, func(i, j int) bool { return CompareVersions(filtered[i].TagName, filtered[j].TagName) > 0 })
|
||||
latest := filtered[0]
|
||||
return &Release{Version: latest.TagName, Name: latest.Name, Body: latest.Body, URL: latest.HTMLURL, PublishedAt: latest.PublishedAt, Prerelease: latest.Prerelease}, nil
|
||||
}
|
||||
|
||||
func (m *Manager) currentVersion() (string, error) {
|
||||
values, err := readEnvFile(m.envPath())
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
version := strings.TrimSpace(values["CANVAS_IMAGE_TAG"])
|
||||
if version == "" || version == "latest" {
|
||||
return "", errors.New("CANVAS_IMAGE_TAG 必须固定为发布版本，不能使用 latest")
|
||||
}
|
||||
if !strings.HasPrefix(version, "v") {
|
||||
version = "v" + version
|
||||
}
|
||||
return version, nil
|
||||
}
|
||||
|
||||
func (m *Manager) prepareTargetCompose(targetVersion string) (string, error) {
|
||||
url := fmt.Sprintf("https://raw.githubusercontent.com/%s/%s/%s", m.config.Repository, targetVersion, m.config.ComposeFile)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
|
||||
defer cancel()
|
||||
request, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
response, err := m.httpClient.Do(request)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("下载目标 Compose：%w", err)
|
||||
}
|
||||
defer response.Body.Close()
|
||||
if response.StatusCode != http.StatusOK {
|
||||
return "", fmt.Errorf("下载目标 Compose 返回 HTTP %d", response.StatusCode)
|
||||
}
|
||||
data, err := io.ReadAll(io.LimitReader(response.Body, 2<<20))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if len(data) == 0 {
|
||||
return "", errors.New("目标 Compose 文件为空")
|
||||
}
|
||||
path := filepath.Join(m.config.StateDir, "compose-"+strings.TrimPrefix(targetVersion, "v")+".next.yml")
|
||||
if err := os.WriteFile(path, data, 0o600); err != nil {
|
||||
return "", fmt.Errorf("保存目标 Compose：%w", err)
|
||||
}
|
||||
return path, nil
|
||||
}
|
||||
|
||||
func (m *Manager) preflight(composePath, targetVersion string) error {
|
||||
if _, err := os.Stat(m.composePath()); err != nil {
|
||||
return fmt.Errorf("读取当前 Compose：%w", err)
|
||||
}
|
||||
if _, err := os.Stat(m.envPath()); err != nil {
|
||||
return fmt.Errorf("读取部署环境：%w", err)
|
||||
}
|
||||
var disk syscall.Statfs_t
|
||||
if err := syscall.Statfs(m.config.BackupDir, &disk); err != nil {
|
||||
return fmt.Errorf("检查备份磁盘空间：%w", err)
|
||||
}
|
||||
available := int64(disk.Bavail) * int64(disk.Bsize)
|
||||
if available < 2<<30 {
|
||||
return fmt.Errorf("备份目录可用空间不足 2 GiB：当前 %d MiB", available>>20)
|
||||
}
|
||||
if err := m.compose(composePath, targetVersion, 2*time.Minute, nil, "config", "--quiet"); err != nil {
|
||||
return fmt.Errorf("目标 Compose 校验失败：%w", err)
|
||||
}
|
||||
current, err := m.currentVersion()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := m.checkHealthOnce(m.healthURL(), current); err != nil {
|
||||
return fmt.Errorf("当前运行版本与部署配置不一致或服务未就绪：%w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) prepareUpdaterBinary(targetVersion string) (string, error) {
|
||||
if !m.config.SelfUpdate {
|
||||
return "", nil
|
||||
}
|
||||
if runtime.GOOS != "linux" || (runtime.GOARCH != "amd64" && runtime.GOARCH != "arm64") {
|
||||
return "", fmt.Errorf("Host Updater 自更新不支持 %s/%s", runtime.GOOS, runtime.GOARCH)
|
||||
}
|
||||
if strings.TrimSpace(m.config.BinaryPath) == "" {
|
||||
return "", errors.New("未配置 Host Updater 二进制路径")
|
||||
}
|
||||
baseURL := fmt.Sprintf("https://github.com/%s/releases/download/%s/", m.config.Repository, targetVersion)
|
||||
asset := "open-ai-canvas-host-updater-linux-" + runtime.GOARCH
|
||||
checksums, err := m.downloadReleaseAsset(baseURL+"SHA256SUMS", 1<<20)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("下载 Host Updater 校验清单：%w", err)
|
||||
}
|
||||
expected := ""
|
||||
for _, line := range strings.Split(string(checksums), "\n") {
|
||||
fields := strings.Fields(line)
|
||||
if len(fields) == 2 && fields[1] == asset {
|
||||
expected = fields[0]
|
||||
break
|
||||
}
|
||||
}
|
||||
if len(expected) != 64 {
|
||||
return "", fmt.Errorf("目标 Release 的 SHA256SUMS 缺少 %s", asset)
|
||||
}
|
||||
binary, err := m.downloadReleaseAsset(baseURL+asset, 128<<20)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("下载目标 Host Updater：%w", err)
|
||||
}
|
||||
hash := sha256.Sum256(binary)
|
||||
actual := hex.EncodeToString(hash[:])
|
||||
if actual != expected {
|
||||
return "", fmt.Errorf("Host Updater 校验失败：期望 %s，实际 %s", expected, actual)
|
||||
}
|
||||
path := filepath.Join(m.config.StateDir, asset+"-"+strings.TrimPrefix(targetVersion, "v")+".next")
|
||||
if err := os.WriteFile(path, binary, 0o700); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return path, nil
|
||||
}
|
||||
|
||||
func (m *Manager) downloadReleaseAsset(url string, limit int64) ([]byte, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||||
defer cancel()
|
||||
request, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
request.Header.Set("User-Agent", "open-ai-canvas-host-updater")
|
||||
response, err := m.httpClient.Do(request)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer response.Body.Close()
|
||||
if response.StatusCode != http.StatusOK {
|
||||
return nil, fmt.Errorf("HTTP %d", response.StatusCode)
|
||||
}
|
||||
reader := io.LimitReader(response.Body, limit+1)
|
||||
data, err := io.ReadAll(reader)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if int64(len(data)) > limit {
|
||||
return nil, errors.New("Release 资产超过允许大小")
|
||||
}
|
||||
return data, nil
|
||||
}
|
||||
|
||||
func (m *Manager) restartSelf() {
|
||||
time.Sleep(time.Second)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
||||
defer cancel()
|
||||
_ = m.runner.Run(ctx, "systemctl", []string{"restart", m.config.ServiceName}, nil, io.Discard, io.Discard)
|
||||
}
|
||||
|
||||
func (m *Manager) compose(composePath, imageTag string, timeout time.Duration, stdout io.Writer, arguments ...string) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer cancel()
|
||||
args := []string{"compose", "--env-file", m.envPath(), "-f", composePath}
|
||||
args = append(args, arguments...)
|
||||
var stderr bytes.Buffer
|
||||
if stdout == nil {
|
||||
stdout = io.Discard
|
||||
}
|
||||
err := m.runner.Run(ctx, "docker", args, []string{"CANVAS_IMAGE_TAG=" + strings.TrimPrefix(imageTag, "v")}, stdout, &stderr)
|
||||
if err != nil {
|
||||
message := strings.TrimSpace(stderr.String())
|
||||
if len(message) > 1000 {
|
||||
message = message[len(message)-1000:]
|
||||
}
|
||||
if message != "" {
|
||||
return fmt.Errorf("docker compose %s：%s", strings.Join(arguments, " "), message)
|
||||
}
|
||||
return fmt.Errorf("docker compose %s：%w", strings.Join(arguments, " "), err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) verifyImages(targetVersion string) error {
|
||||
owner := strings.SplitN(m.config.Repository, "/", 2)[0]
|
||||
for _, image := range []string{"ghcr.io/" + owner + "/open-ai-canvas-backend:", "ghcr.io/" + owner + "/open-ai-canvas-web:"} {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||||
var output bytes.Buffer
|
||||
var stderr bytes.Buffer
|
||||
err := m.runner.Run(ctx, "docker", []string{"image", "inspect", image + strings.TrimPrefix(targetVersion, "v"), "--format", "{{json .RepoDigests}}"}, nil, &output, &stderr)
|
||||
cancel()
|
||||
if err != nil {
|
||||
return fmt.Errorf("校验目标镜像摘要失败：%s", strings.TrimSpace(stderr.String()))
|
||||
}
|
||||
if !strings.Contains(output.String(), "@sha256:") {
|
||||
return fmt.Errorf("目标镜像 %s 未包含仓库摘要", image+targetVersion)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) createBackup(version string) (Backup, error) {
|
||||
now := time.Now().UTC()
|
||||
id := "backup-" + now.Format("20060102-150405")
|
||||
path := filepath.Join(m.config.BackupDir, id+".zip")
|
||||
temporary := path + ".partial"
|
||||
file, err := os.OpenFile(temporary, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
|
||||
if err != nil {
|
||||
return Backup{}, fmt.Errorf("创建备份文件：%w", err)
|
||||
}
|
||||
removeTemporary := true
|
||||
defer func() {
|
||||
_ = file.Close()
|
||||
if removeTemporary {
|
||||
_ = os.Remove(temporary)
|
||||
}
|
||||
}()
|
||||
hasher := sha256.New()
|
||||
archive := zip.NewWriter(io.MultiWriter(file, hasher))
|
||||
metadata, _ := json.MarshalIndent(map[string]any{"id": id, "version": version, "createdAt": now, "format": 1}, "", " ")
|
||||
if err := writeZipBytes(archive, "metadata.json", metadata); err != nil {
|
||||
return Backup{}, err
|
||||
}
|
||||
databaseEntry, err := archive.CreateHeader(&zip.FileHeader{Name: "database.dump", Method: zip.Store})
|
||||
if err != nil {
|
||||
return Backup{}, err
|
||||
}
|
||||
values, err := readEnvFile(m.envPath())
|
||||
if err != nil {
|
||||
return Backup{}, err
|
||||
}
|
||||
postgresUser := firstNonEmpty(values["POSTGRES_USER"], "open_ai_canvas")
|
||||
postgresDB := firstNonEmpty(values["POSTGRES_DB"], "open_ai_canvas")
|
||||
if err := m.compose(m.composePath(), version, m.config.StepTimeout, databaseEntry, "exec", "-T", "postgres", "pg_dump", "-U", postgresUser, "-d", postgresDB, "-Fc"); err != nil {
|
||||
return Backup{}, fmt.Errorf("备份 PostgreSQL：%w", err)
|
||||
}
|
||||
dataEntry, err := archive.CreateHeader(&zip.FileHeader{Name: "backend-data.tar", Method: zip.Store})
|
||||
if err != nil {
|
||||
return Backup{}, err
|
||||
}
|
||||
if err := m.compose(m.composePath(), version, m.config.StepTimeout, dataEntry, "exec", "-T", "backend", "tar", "-C", "/data", "-cf", "-", "."); err != nil {
|
||||
return Backup{}, fmt.Errorf("备份数据目录：%w", err)
|
||||
}
|
||||
if err := archive.Close(); err != nil {
|
||||
return Backup{}, fmt.Errorf("完成 ZIP 备份：%w", err)
|
||||
}
|
||||
if err := file.Sync(); err != nil {
|
||||
return Backup{}, fmt.Errorf("同步 ZIP 备份：%w", err)
|
||||
}
|
||||
if err := file.Close(); err != nil {
|
||||
return Backup{}, err
|
||||
}
|
||||
if err := os.Rename(temporary, path); err != nil {
|
||||
return Backup{}, fmt.Errorf("提交 ZIP 备份：%w", err)
|
||||
}
|
||||
removeTemporary = false
|
||||
stat, err := os.Stat(path)
|
||||
if err != nil {
|
||||
return Backup{}, err
|
||||
}
|
||||
checksum := "sha256:" + hex.EncodeToString(hasher.Sum(nil))
|
||||
if err := verifyZipBackup(path, checksum); err != nil {
|
||||
return Backup{}, err
|
||||
}
|
||||
return Backup{ID: id, Path: path, Checksum: checksum, Size: stat.Size(), CreatedAt: now, Version: version}, nil
|
||||
}
|
||||
|
||||
func writeZipBytes(writer *zip.Writer, name string, data []byte) error {
|
||||
entry, err := writer.CreateHeader(&zip.FileHeader{Name: name, Method: zip.Deflate})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = entry.Write(data)
|
||||
return err
|
||||
}
|
||||
|
||||
func verifyZipBackup(path, expected string) error {
|
||||
file, err := os.Open(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
hasher := sha256.New()
|
||||
if _, err := io.Copy(hasher, file); err != nil {
|
||||
file.Close()
|
||||
return err
|
||||
}
|
||||
if err := file.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
actual := "sha256:" + hex.EncodeToString(hasher.Sum(nil))
|
||||
if actual != expected {
|
||||
return fmt.Errorf("ZIP 备份校验失败：期望 %s，实际 %s", expected, actual)
|
||||
}
|
||||
archive, err := zip.OpenReader(path)
|
||||
if err != nil {
|
||||
return fmt.Errorf("ZIP 备份无法打开：%w", err)
|
||||
}
|
||||
defer archive.Close()
|
||||
required := map[string]bool{"metadata.json": false, "database.dump": false, "backend-data.tar": false}
|
||||
for _, entry := range archive.File {
|
||||
if _, ok := required[entry.Name]; ok && entry.UncompressedSize64 > 0 {
|
||||
required[entry.Name] = true
|
||||
}
|
||||
}
|
||||
for name, present := range required {
|
||||
if !present {
|
||||
return fmt.Errorf("ZIP 备份缺少有效文件 %s", name)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) restoreDatabase(backup Backup) error {
|
||||
if err := verifyZipBackup(backup.Path, backup.Checksum); err != nil {
|
||||
return fmt.Errorf("拒绝恢复未通过校验的备份：%w", err)
|
||||
}
|
||||
archive, err := zip.OpenReader(backup.Path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer archive.Close()
|
||||
var dump *zip.File
|
||||
for _, entry := range archive.File {
|
||||
if entry.Name == "database.dump" {
|
||||
dump = entry
|
||||
break
|
||||
}
|
||||
}
|
||||
if dump == nil {
|
||||
return errors.New("备份中不存在 database.dump")
|
||||
}
|
||||
reader, err := dump.Open()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer reader.Close()
|
||||
values, err := readEnvFile(m.envPath())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
postgresUser := firstNonEmpty(values["POSTGRES_USER"], "open_ai_canvas")
|
||||
postgresDB := firstNonEmpty(values["POSTGRES_DB"], "open_ai_canvas")
|
||||
if postgresDB == "postgres" || strings.HasPrefix(postgresDB, "template") {
|
||||
return fmt.Errorf("拒绝覆盖 PostgreSQL 系统数据库 %q", postgresDB)
|
||||
}
|
||||
if err := m.compose(m.composePath(), backup.Version, 5*time.Minute, nil, "exec", "-T", "postgres", "dropdb", "-U", postgresUser, "--if-exists", "--force", postgresDB); err != nil {
|
||||
return fmt.Errorf("删除待恢复数据库：%w", err)
|
||||
}
|
||||
if err := m.compose(m.composePath(), backup.Version, 5*time.Minute, nil, "exec", "-T", "postgres", "createdb", "-U", postgresUser, "-O", postgresUser, postgresDB); err != nil {
|
||||
return fmt.Errorf("重建待恢复数据库：%w", err)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), m.config.StepTimeout)
|
||||
defer cancel()
|
||||
args := []string{"compose", "--env-file", m.envPath(), "-f", m.composePath(), "exec", "-T", "postgres", "pg_restore", "-U", postgresUser, "-d", postgresDB, "--no-owner", "--no-privileges"}
|
||||
var stderr bytes.Buffer
|
||||
command := execCommandWithInput{runner: m.runner, input: reader}
|
||||
if err := command.Run(ctx, "docker", args, []string{"CANVAS_IMAGE_TAG=" + strings.TrimPrefix(backup.Version, "v")}, io.Discard, &stderr); err != nil {
|
||||
return fmt.Errorf("恢复 PostgreSQL：%s", strings.TrimSpace(stderr.String()))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type execCommandWithInput struct {
|
||||
runner commandRunner
|
||||
input io.Reader
|
||||
}
|
||||
|
||||
func (c execCommandWithInput) Run(ctx context.Context, name string, args, environment []string, stdout, stderr io.Writer) error {
|
||||
runner, ok := c.runner.(execRunner)
|
||||
if !ok {
|
||||
return errors.New("当前命令执行器不支持标准输入")
|
||||
}
|
||||
command := exec.CommandContext(ctx, name, args...)
|
||||
command.Dir = runner.dir
|
||||
command.Env = append(os.Environ(), environment...)
|
||||
command.Stdin = c.input
|
||||
command.Stdout = stdout
|
||||
command.Stderr = stderr
|
||||
return command.Run()
|
||||
}
|
||||
|
||||
func (m *Manager) verifyHealthy(targetVersion string) error {
|
||||
healthURL := m.healthURL()
|
||||
deadline := time.Now().Add(10 * time.Minute)
|
||||
stableSince := time.Time{}
|
||||
for time.Now().Before(deadline) {
|
||||
if err := m.checkHealthOnce(healthURL, targetVersion); err != nil {
|
||||
stableSince = time.Time{}
|
||||
} else if stableSince.IsZero() {
|
||||
stableSince = time.Now()
|
||||
} else if time.Since(stableSince) >= m.config.StableWindow {
|
||||
return nil
|
||||
}
|
||||
time.Sleep(2 * time.Second)
|
||||
}
|
||||
return fmt.Errorf("目标版本 %s 未在健康检查窗口内稳定就绪", targetVersion)
|
||||
}
|
||||
|
||||
func (m *Manager) healthURL() string {
|
||||
if configured := strings.TrimSpace(m.config.HealthURL); configured != "" {
|
||||
return configured
|
||||
}
|
||||
values, err := readEnvFile(m.envPath())
|
||||
if err != nil {
|
||||
return "http://127.0.0.1:3000/api/health/ready"
|
||||
}
|
||||
return "http://127.0.0.1:" + firstNonEmpty(values["CANVAS_HTTP_PORT"], "3000") + "/api/health/ready"
|
||||
}
|
||||
|
||||
func (m *Manager) checkHealthOnce(healthURL, targetVersion string) error {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
request, _ := http.NewRequestWithContext(ctx, http.MethodGet, healthURL, nil)
|
||||
response, err := m.httpClient.Do(request)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer response.Body.Close()
|
||||
if response.StatusCode != http.StatusOK {
|
||||
return fmt.Errorf("健康接口返回 HTTP %d", response.StatusCode)
|
||||
}
|
||||
var payload struct {
|
||||
Code int `json:"code"`
|
||||
Data struct {
|
||||
Build struct {
|
||||
Version string `json:"version"`
|
||||
} `json:"build"`
|
||||
} `json:"data"`
|
||||
}
|
||||
if err := json.NewDecoder(io.LimitReader(response.Body, 256<<10)).Decode(&payload); err != nil {
|
||||
return err
|
||||
}
|
||||
if payload.Code != 0 {
|
||||
return errors.New("健康接口业务状态异常")
|
||||
}
|
||||
if payload.Data.Build.Version != "" && CompareVersions(payload.Data.Build.Version, targetVersion) != 0 {
|
||||
return fmt.Errorf("运行版本仍为 %s，期望 %s", payload.Data.Build.Version, targetVersion)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func readEnvFile(path string) (map[string]string, error) {
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("读取 %s：%w", path, err)
|
||||
}
|
||||
values := make(map[string]string)
|
||||
for _, line := range strings.Split(string(data), "\n") {
|
||||
line = strings.TrimSpace(line)
|
||||
if line == "" || strings.HasPrefix(line, "#") {
|
||||
continue
|
||||
}
|
||||
key, value, ok := strings.Cut(line, "=")
|
||||
if ok {
|
||||
values[strings.TrimSpace(key)] = strings.Trim(strings.TrimSpace(value), "\"")
|
||||
}
|
||||
}
|
||||
return values, nil
|
||||
}
|
||||
|
||||
func setEnvValue(path, key, value string) error {
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
lines := strings.Split(strings.TrimSuffix(string(data), "\n"), "\n")
|
||||
found := false
|
||||
for index, line := range lines {
|
||||
trimmed := strings.TrimSpace(line)
|
||||
if strings.HasPrefix(trimmed, key+"=") {
|
||||
lines[index] = key + "=" + value
|
||||
found = true
|
||||
}
|
||||
}
|
||||
if !found {
|
||||
lines = append(lines, key+"="+value)
|
||||
}
|
||||
stat, err := os.Stat(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
temporary, err := os.CreateTemp(filepath.Dir(path), ".env-*.tmp")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
temporaryName := temporary.Name()
|
||||
defer os.Remove(temporaryName)
|
||||
if err := temporary.Chmod(stat.Mode().Perm()); err != nil {
|
||||
temporary.Close()
|
||||
return err
|
||||
}
|
||||
if _, err := temporary.WriteString(strings.Join(lines, "\n") + "\n"); err != nil {
|
||||
temporary.Close()
|
||||
return err
|
||||
}
|
||||
if err := temporary.Sync(); err != nil {
|
||||
temporary.Close()
|
||||
return err
|
||||
}
|
||||
if err := temporary.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.Rename(temporaryName, path)
|
||||
}
|
||||
|
||||
func replaceFile(source, target string, mode os.FileMode) error {
|
||||
data, err := os.ReadFile(source)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if mode == 0 {
|
||||
if stat, statErr := os.Stat(target); statErr == nil {
|
||||
mode = stat.Mode().Perm()
|
||||
} else {
|
||||
mode = 0o600
|
||||
}
|
||||
}
|
||||
temporary, err := os.CreateTemp(filepath.Dir(target), ".compose-*.tmp")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
temporaryName := temporary.Name()
|
||||
defer os.Remove(temporaryName)
|
||||
if err := temporary.Chmod(mode); err != nil {
|
||||
temporary.Close()
|
||||
return err
|
||||
}
|
||||
if _, err := temporary.Write(data); err != nil {
|
||||
temporary.Close()
|
||||
return err
|
||||
}
|
||||
if err := temporary.Sync(); err != nil {
|
||||
temporary.Close()
|
||||
return err
|
||||
}
|
||||
if err := temporary.Close(); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.Rename(temporaryName, target)
|
||||
}
|
||||
|
||||
func firstNonEmpty(values ...string) string {
|
||||
for _, value := range values {
|
||||
if strings.TrimSpace(value) != "" {
|
||||
return strings.TrimSpace(value)
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
92
backend/internal/hostupdate/manager_ops_test.go
Normal file
92
backend/internal/hostupdate/manager_ops_test.go
Normal file
@@ -0,0 +1,92 @@
|
||||
package hostupdate
|
||||
|
||||
import (
|
||||
"archive/zip"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestSetEnvValuePreservesOtherSettings(t *testing.T) {
|
||||
directory := t.TempDir()
|
||||
path := filepath.Join(directory, ".env")
|
||||
if err := os.WriteFile(path, []byte("# keep\nCANVAS_IMAGE_TAG=1.0.0\nPOSTGRES_DB=canvas\n"), 0o640); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := setEnvValue(path, "CANVAS_IMAGE_TAG", "1.2.2-preview.1"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
value := string(data)
|
||||
if !strings.Contains(value, "# keep\n") || !strings.Contains(value, "POSTGRES_DB=canvas\n") || !strings.Contains(value, "CANVAS_IMAGE_TAG=1.2.2-preview.1\n") {
|
||||
t.Fatalf("unexpected env contents: %q", value)
|
||||
}
|
||||
stat, err := os.Stat(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if stat.Mode().Perm() != 0o640 {
|
||||
t.Fatalf("mode=%o, want 640", stat.Mode().Perm())
|
||||
}
|
||||
}
|
||||
|
||||
func TestVerifyZipBackupRejectsCorruption(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "backup.zip")
|
||||
file, err := os.Create(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
archive := zip.NewWriter(file)
|
||||
for name, content := range map[string]string{
|
||||
"metadata.json": "{}",
|
||||
"database.dump": "database",
|
||||
"backend-data.tar": "data",
|
||||
} {
|
||||
entry, createErr := archive.Create(name)
|
||||
if createErr != nil {
|
||||
t.Fatal(createErr)
|
||||
}
|
||||
if _, writeErr := io.WriteString(entry, content); writeErr != nil {
|
||||
t.Fatal(writeErr)
|
||||
}
|
||||
}
|
||||
if err := archive.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := file.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
hash := sha256.Sum256(data)
|
||||
checksum := "sha256:" + hex.EncodeToString(hash[:])
|
||||
if err := verifyZipBackup(path, checksum); err != nil {
|
||||
t.Fatalf("valid backup rejected: %v", err)
|
||||
}
|
||||
if err := os.WriteFile(path, append(data, byte(1)), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := verifyZipBackup(path, checksum); err == nil {
|
||||
t.Fatal("corrupted backup was accepted")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCurrentVersionRejectsLatest(t *testing.T) {
|
||||
installDir := t.TempDir()
|
||||
if err := os.WriteFile(filepath.Join(installDir, ".env"), []byte("CANVAS_IMAGE_TAG=latest\n"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
manager := &Manager{config: Config{InstallDir: installDir, EnvFile: ".env"}}
|
||||
if _, err := manager.currentVersion(); err == nil {
|
||||
t.Fatal("latest tag was accepted")
|
||||
}
|
||||
}
|
||||
101
backend/internal/hostupdate/server.go
Normal file
101
backend/internal/hostupdate/server.go
Normal file
@@ -0,0 +1,101 @@
|
||||
package hostupdate
|
||||
|
||||
import (
|
||||
"crypto/subtle"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
)
|
||||
|
||||
type Server struct {
|
||||
manager *Manager
|
||||
token string
|
||||
}
|
||||
|
||||
func NewServer(manager *Manager, token string) (*Server, error) {
|
||||
if manager == nil {
|
||||
return nil, errors.New("更新管理器不能为空")
|
||||
}
|
||||
if len(strings.TrimSpace(token)) < 32 {
|
||||
return nil, errors.New("CANVAS_UPDATER_TOKEN 至少需要 32 个字符")
|
||||
}
|
||||
return &Server{manager: manager, token: strings.TrimSpace(token)}, nil
|
||||
}
|
||||
|
||||
func (s *Server) Handler() http.Handler {
|
||||
mux := http.NewServeMux()
|
||||
mux.HandleFunc("GET /v1/status", func(w http.ResponseWriter, _ *http.Request) { writeJSON(w, http.StatusOK, s.manager.Snapshot()) })
|
||||
mux.HandleFunc("POST /v1/check", func(w http.ResponseWriter, r *http.Request) {
|
||||
status, err := s.manager.Check(r.Context())
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadGateway, err, status)
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, status)
|
||||
})
|
||||
mux.HandleFunc("POST /v1/update", func(w http.ResponseWriter, r *http.Request) {
|
||||
var request StartRequest
|
||||
if err := decodeJSON(r, &request); err != nil {
|
||||
writeError(w, http.StatusBadRequest, err, nil)
|
||||
return
|
||||
}
|
||||
status, err := s.manager.StartUpdate(request.TargetVersion)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusConflict, err, status)
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusAccepted, status)
|
||||
})
|
||||
mux.HandleFunc("POST /v1/rollback", func(w http.ResponseWriter, r *http.Request) {
|
||||
var request RollbackRequest
|
||||
if err := decodeJSON(r, &request); err != nil {
|
||||
writeError(w, http.StatusBadRequest, err, nil)
|
||||
return
|
||||
}
|
||||
status, err := s.manager.StartRollback(request.Reason)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusConflict, err, status)
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusAccepted, status)
|
||||
})
|
||||
return s.authorize(mux)
|
||||
}
|
||||
|
||||
func (s *Server) authorize(next http.Handler) http.Handler {
|
||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
provided := strings.TrimSpace(strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer "))
|
||||
if len(provided) != len(s.token) || subtle.ConstantTimeCompare([]byte(provided), []byte(s.token)) != 1 {
|
||||
writeError(w, http.StatusUnauthorized, errors.New("更新器认证失败"), nil)
|
||||
return
|
||||
}
|
||||
w.Header().Set("Cache-Control", "no-store")
|
||||
next.ServeHTTP(w, r)
|
||||
})
|
||||
}
|
||||
|
||||
func decodeJSON(r *http.Request, target any) error {
|
||||
defer r.Body.Close()
|
||||
decoder := json.NewDecoder(io.LimitReader(r.Body, 64<<10))
|
||||
decoder.DisallowUnknownFields()
|
||||
if err := decoder.Decode(target); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func writeError(w http.ResponseWriter, status int, err error, data any) {
|
||||
writeJSONStatus(w, status, map[string]any{"error": err.Error(), "data": data})
|
||||
}
|
||||
|
||||
func writeJSON(w http.ResponseWriter, status int, value any) {
|
||||
writeJSONStatus(w, status, value)
|
||||
}
|
||||
|
||||
func writeJSONStatus(w http.ResponseWriter, status int, value any) {
|
||||
w.Header().Set("Content-Type", "application/json; charset=utf-8")
|
||||
w.WriteHeader(status)
|
||||
_ = json.NewEncoder(w).Encode(value)
|
||||
}
|
||||
100
backend/internal/hostupdate/types.go
Normal file
100
backend/internal/hostupdate/types.go
Normal file
@@ -0,0 +1,100 @@
|
||||
package hostupdate
|
||||
|
||||
import "time"
|
||||
|
||||
type Phase string
|
||||
|
||||
const (
|
||||
PhaseIdle Phase = "idle"
|
||||
PhaseChecking Phase = "checking"
|
||||
PhaseReady Phase = "ready"
|
||||
PhaseNoUpdate Phase = "no_update"
|
||||
PhasePreflight Phase = "preflight"
|
||||
PhaseBackingUp Phase = "backing_up"
|
||||
PhasePulling Phase = "pulling"
|
||||
PhaseDraining Phase = "draining"
|
||||
PhaseMigrating Phase = "migrating"
|
||||
PhaseSwitching Phase = "switching"
|
||||
PhaseVerifying Phase = "verifying"
|
||||
PhaseSucceeded Phase = "succeeded"
|
||||
PhaseRollingBack Phase = "rolling_back"
|
||||
PhaseRolledBack Phase = "rolled_back"
|
||||
PhaseFailed Phase = "failed"
|
||||
PhaseManualIntervention Phase = "manual_intervention"
|
||||
)
|
||||
|
||||
func (p Phase) Active() bool {
|
||||
switch p {
|
||||
case PhasePreflight, PhaseBackingUp, PhasePulling, PhaseDraining, PhaseMigrating, PhaseSwitching, PhaseVerifying, PhaseRollingBack:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
type Release struct {
|
||||
Version string `json:"version"`
|
||||
Name string `json:"name"`
|
||||
Body string `json:"body"`
|
||||
URL string `json:"url"`
|
||||
PublishedAt time.Time `json:"publishedAt"`
|
||||
Prerelease bool `json:"prerelease"`
|
||||
}
|
||||
|
||||
type Check struct {
|
||||
Key string `json:"key"`
|
||||
Label string `json:"label"`
|
||||
Status string `json:"status"`
|
||||
Detail string `json:"detail,omitempty"`
|
||||
Blocking bool `json:"blocking"`
|
||||
}
|
||||
|
||||
type Backup struct {
|
||||
ID string `json:"id"`
|
||||
Path string `json:"path"`
|
||||
Checksum string `json:"checksum"`
|
||||
Size int64 `json:"size"`
|
||||
CreatedAt time.Time `json:"createdAt"`
|
||||
Version string `json:"version"`
|
||||
}
|
||||
|
||||
type LogEntry struct {
|
||||
At time.Time `json:"at"`
|
||||
Phase Phase `json:"phase"`
|
||||
Message string `json:"message"`
|
||||
}
|
||||
|
||||
type Operation struct {
|
||||
ID string `json:"id,omitempty"`
|
||||
Phase Phase `json:"phase"`
|
||||
FromVersion string `json:"fromVersion,omitempty"`
|
||||
TargetVersion string `json:"targetVersion,omitempty"`
|
||||
StartedAt *time.Time `json:"startedAt,omitempty"`
|
||||
FinishedAt *time.Time `json:"finishedAt,omitempty"`
|
||||
Error string `json:"error,omitempty"`
|
||||
RollbackError string `json:"rollbackError,omitempty"`
|
||||
AutomaticRollback bool `json:"automaticRollback"`
|
||||
Logs []LogEntry `json:"logs"`
|
||||
}
|
||||
|
||||
type Status struct {
|
||||
Supported bool `json:"supported"`
|
||||
Connected bool `json:"connected"`
|
||||
Repository string `json:"repository"`
|
||||
Deployment string `json:"deployment"`
|
||||
CurrentVersion string `json:"currentVersion"`
|
||||
LatestRelease *Release `json:"latestRelease,omitempty"`
|
||||
UpdateAvailable bool `json:"updateAvailable"`
|
||||
Checks []Check `json:"checks"`
|
||||
LastBackup *Backup `json:"lastBackup,omitempty"`
|
||||
RollbackVersion string `json:"rollbackVersion,omitempty"`
|
||||
Operation Operation `json:"operation"`
|
||||
}
|
||||
|
||||
type StartRequest struct {
|
||||
TargetVersion string `json:"targetVersion"`
|
||||
}
|
||||
|
||||
type RollbackRequest struct {
|
||||
Reason string `json:"reason"`
|
||||
}
|
||||
79
backend/internal/hostupdate/version.go
Normal file
79
backend/internal/hostupdate/version.go
Normal file
@@ -0,0 +1,79 @@
|
||||
package hostupdate
|
||||
|
||||
import (
|
||||
"strconv"
|
||||
"strings"
|
||||
)
|
||||
|
||||
type versionPart struct {
|
||||
major, minor, patch int
|
||||
pre []string
|
||||
}
|
||||
|
||||
func CompareVersions(left, right string) int {
|
||||
a, aok := parseVersion(left)
|
||||
b, bok := parseVersion(right)
|
||||
if !aok || !bok {
|
||||
return strings.Compare(strings.TrimSpace(left), strings.TrimSpace(right))
|
||||
}
|
||||
for _, pair := range [][2]int{{a.major, b.major}, {a.minor, b.minor}, {a.patch, b.patch}} {
|
||||
if pair[0] < pair[1] {
|
||||
return -1
|
||||
}
|
||||
if pair[0] > pair[1] {
|
||||
return 1
|
||||
}
|
||||
}
|
||||
if len(a.pre) == 0 && len(b.pre) > 0 {
|
||||
return 1
|
||||
}
|
||||
if len(a.pre) > 0 && len(b.pre) == 0 {
|
||||
return -1
|
||||
}
|
||||
for index := 0; index < len(a.pre) || index < len(b.pre); index++ {
|
||||
if index >= len(a.pre) {
|
||||
return -1
|
||||
}
|
||||
if index >= len(b.pre) {
|
||||
return 1
|
||||
}
|
||||
an, aerr := strconv.Atoi(a.pre[index])
|
||||
bn, berr := strconv.Atoi(b.pre[index])
|
||||
switch {
|
||||
case aerr == nil && berr == nil && an < bn:
|
||||
return -1
|
||||
case aerr == nil && berr == nil && an > bn:
|
||||
return 1
|
||||
case aerr == nil && berr != nil:
|
||||
return -1
|
||||
case aerr != nil && berr == nil:
|
||||
return 1
|
||||
case a.pre[index] < b.pre[index]:
|
||||
return -1
|
||||
case a.pre[index] > b.pre[index]:
|
||||
return 1
|
||||
}
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func parseVersion(raw string) (versionPart, bool) {
|
||||
value := strings.TrimPrefix(strings.TrimSpace(raw), "v")
|
||||
value = strings.SplitN(value, "+", 2)[0]
|
||||
parts := strings.SplitN(value, "-", 2)
|
||||
core := strings.Split(parts[0], ".")
|
||||
if len(core) != 3 {
|
||||
return versionPart{}, false
|
||||
}
|
||||
major, err1 := strconv.Atoi(core[0])
|
||||
minor, err2 := strconv.Atoi(core[1])
|
||||
patch, err3 := strconv.Atoi(core[2])
|
||||
if err1 != nil || err2 != nil || err3 != nil {
|
||||
return versionPart{}, false
|
||||
}
|
||||
parsed := versionPart{major: major, minor: minor, patch: patch}
|
||||
if len(parts) == 2 && parts[1] != "" {
|
||||
parsed.pre = strings.FieldsFunc(parts[1], func(r rune) bool { return r == '.' || r == '-' })
|
||||
}
|
||||
return parsed, true
|
||||
}
|
||||
27
backend/internal/hostupdate/version_test.go
Normal file
27
backend/internal/hostupdate/version_test.go
Normal file
@@ -0,0 +1,27 @@
|
||||
package hostupdate
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestCompareVersions(t *testing.T) {
|
||||
tests := []struct {
|
||||
left, right string
|
||||
want int
|
||||
}{
|
||||
{"v1.2.0-preview.2", "v1.2.0-preview.3", -1},
|
||||
{"v1.2.0-preview.10", "v1.2.0-preview.3", 1},
|
||||
{"v1.2.0-preview.3", "v1.2.0", -1},
|
||||
{"v1.2.1", "v1.2.0", 1},
|
||||
{"v1.2.0", "v1.2.0", 0},
|
||||
}
|
||||
for _, test := range tests {
|
||||
got := CompareVersions(test.left, test.right)
|
||||
if got < 0 {
|
||||
got = -1
|
||||
} else if got > 0 {
|
||||
got = 1
|
||||
}
|
||||
if got != test.want {
|
||||
t.Fatalf("CompareVersions(%q, %q)=%d, want %d", test.left, test.right, got, test.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -56,6 +56,12 @@ func (r *Repository) Dialect() string {
|
||||
return r.db.Dialector.Name()
|
||||
}
|
||||
|
||||
func (r *Repository) ReleaseTaskLease(id string, owner string) error {
|
||||
return r.db.Model(&model.Task{}).
|
||||
Where("id = ? AND status = ? AND lease_owner = ?", id, model.TaskStatusRunning, owner).
|
||||
Updates(map[string]any{"lease_owner": "", "lease_expires_at": nil, "updated_at": time.Now()}).Error
|
||||
}
|
||||
|
||||
// NextPrefixedID 在数据库事务中递增序列，避免 UUID/父子字符串拼接导致的不可读和不可排序 ID。
|
||||
// prefix 只决定展示前缀，关联关系仍由独立外键维护。
|
||||
func (r *Repository) NextPrefixedID(prefix string) (string, error) {
|
||||
|
||||
94
backend/internal/service/admin_update.go
Normal file
94
backend/internal/service/admin_update.go
Normal file
@@ -0,0 +1,94 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"infinite-canvas/backend/internal/hostupdate"
|
||||
"infinite-canvas/backend/internal/model"
|
||||
)
|
||||
|
||||
type UpdateManager interface {
|
||||
Status(context.Context) (hostupdate.Status, error)
|
||||
Check(context.Context) (hostupdate.Status, error)
|
||||
Start(context.Context, string) (hostupdate.Status, error)
|
||||
Rollback(context.Context, string) (hostupdate.Status, error)
|
||||
}
|
||||
|
||||
func (s *Service) ConfigureUpdateManager(manager UpdateManager) {
|
||||
s.updateManager = manager
|
||||
}
|
||||
|
||||
func (s *Service) AdminUpdateStatus(ctx context.Context, actor *model.User) (hostupdate.Status, error) {
|
||||
if err := s.RequireAdmin(actor); err != nil {
|
||||
return hostupdate.Status{}, err
|
||||
}
|
||||
if s.updateManager == nil {
|
||||
return unsupportedUpdateStatus("当前部署未安装 Host Updater"), nil
|
||||
}
|
||||
status, err := s.updateManager.Status(ctx)
|
||||
if err != nil {
|
||||
return unsupportedUpdateStatus("Host Updater 当前不可连接"), nil
|
||||
}
|
||||
return status, nil
|
||||
}
|
||||
|
||||
func (s *Service) AdminCheckUpdate(ctx context.Context, actor *model.User) (hostupdate.Status, error) {
|
||||
if err := s.RequireAdmin(actor); err != nil {
|
||||
return hostupdate.Status{}, err
|
||||
}
|
||||
if s.updateManager == nil {
|
||||
return hostupdate.Status{}, NewAppError(http.StatusServiceUnavailable, "当前部署未安装 Host Updater")
|
||||
}
|
||||
status, err := s.updateManager.Check(ctx)
|
||||
if err != nil {
|
||||
return status, WrapAppError(http.StatusBadGateway, "检查更新失败，请查看更新器状态和日志", err)
|
||||
}
|
||||
return status, nil
|
||||
}
|
||||
|
||||
func (s *Service) AdminStartUpdate(ctx context.Context, actor *model.User, targetVersion string) (hostupdate.Status, error) {
|
||||
if err := s.RequireAdmin(actor); err != nil {
|
||||
return hostupdate.Status{}, err
|
||||
}
|
||||
if s.updateManager == nil {
|
||||
return hostupdate.Status{}, NewAppError(http.StatusServiceUnavailable, "当前部署未安装 Host Updater")
|
||||
}
|
||||
targetVersion = strings.TrimSpace(targetVersion)
|
||||
if targetVersion == "" {
|
||||
return hostupdate.Status{}, NewAppError(http.StatusBadRequest, "目标版本不能为空")
|
||||
}
|
||||
status, err := s.updateManager.Start(ctx, targetVersion)
|
||||
if err != nil {
|
||||
return status, WrapAppError(http.StatusConflict, "无法开始更新，请刷新状态后重试", err)
|
||||
}
|
||||
return status, nil
|
||||
}
|
||||
|
||||
func (s *Service) AdminRollbackUpdate(ctx context.Context, actor *model.User, reason string) (hostupdate.Status, error) {
|
||||
if err := s.RequireAdmin(actor); err != nil {
|
||||
return hostupdate.Status{}, err
|
||||
}
|
||||
if s.updateManager == nil {
|
||||
return hostupdate.Status{}, NewAppError(http.StatusServiceUnavailable, "当前部署未安装 Host Updater")
|
||||
}
|
||||
if strings.TrimSpace(reason) == "" {
|
||||
return hostupdate.Status{}, NewAppError(http.StatusBadRequest, "请填写回退原因")
|
||||
}
|
||||
status, err := s.updateManager.Rollback(ctx, reason)
|
||||
if err != nil {
|
||||
return status, WrapAppError(http.StatusConflict, "无法开始回退，请检查备份和当前状态", err)
|
||||
}
|
||||
return status, nil
|
||||
}
|
||||
|
||||
func unsupportedUpdateStatus(detail string) hostupdate.Status {
|
||||
return hostupdate.Status{
|
||||
Supported: false,
|
||||
Connected: false,
|
||||
Deployment: "unsupported",
|
||||
Checks: []hostupdate.Check{{Key: "updater", Label: "Host Updater", Status: "failed", Detail: detail, Blocking: true}},
|
||||
Operation: hostupdate.Operation{Phase: hostupdate.PhaseIdle, Logs: []hostupdate.LogEntry{}},
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,7 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"time"
|
||||
)
|
||||
@@ -25,18 +26,23 @@ func (s *Service) AuditBillingReview() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Service) startBillingReviewAudit() {
|
||||
func (s *Service) startBillingReviewAudit(ctx context.Context) {
|
||||
audit := func() {
|
||||
if err := s.AuditBillingReview(); err != nil {
|
||||
log.Printf("billing review audit failed: %v", err)
|
||||
}
|
||||
}
|
||||
audit()
|
||||
go func() {
|
||||
s.runWorkerLoop(func(ctx context.Context) {
|
||||
audit()
|
||||
ticker := time.NewTicker(time.Hour)
|
||||
defer ticker.Stop()
|
||||
for range ticker.C {
|
||||
audit()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
audit()
|
||||
}
|
||||
}
|
||||
}()
|
||||
})
|
||||
}
|
||||
|
||||
@@ -317,6 +317,13 @@ func (s *Service) ValidateRuntime() error {
|
||||
return s.runtimeErr
|
||||
}
|
||||
|
||||
func (s *Service) Close() error {
|
||||
if s.coordinator == nil || s.coordinator.redis == nil {
|
||||
return nil
|
||||
}
|
||||
return s.coordinator.redis.Close()
|
||||
}
|
||||
|
||||
func (s *Service) AllowRequest(ctx context.Context, key string, limit int, window time.Duration) (bool, error) {
|
||||
if s.coordinator == nil {
|
||||
return false, errors.New("运行时协调器未初始化")
|
||||
|
||||
@@ -31,22 +31,27 @@ const (
|
||||
providerCancellationFailed providerCancellationOutcome = "failed"
|
||||
)
|
||||
|
||||
func (s *Service) startProviderCancellationReconciliation() {
|
||||
go func() {
|
||||
func (s *Service) startProviderCancellationReconciliation(ctx context.Context) {
|
||||
s.runWorkerLoop(func(ctx context.Context) {
|
||||
ticker := time.NewTicker(5 * time.Second)
|
||||
defer ticker.Stop()
|
||||
for range ticker.C {
|
||||
task, err := s.repo.ClaimNextTaskProviderCancellation("provider-cancel:"+s.workerID, providerCancellationLeaseDuration)
|
||||
if err != nil || task == nil {
|
||||
continue
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
task, err := s.repo.ClaimNextTaskProviderCancellation("provider-cancel:"+s.workerID, providerCancellationLeaseDuration)
|
||||
if err != nil || task == nil {
|
||||
continue
|
||||
}
|
||||
reconcileCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
|
||||
if err := s.reconcileProviderCancellation(reconcileCtx, task); err != nil {
|
||||
_ = s.log(task.UserID, task.ID, "error", "上游取消状态对账失败", err.Error())
|
||||
}
|
||||
cancel()
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
if err := s.reconcileProviderCancellation(ctx, task); err != nil {
|
||||
_ = s.log(task.UserID, task.ID, "error", "上游取消状态对账失败", err.Error())
|
||||
}
|
||||
cancel()
|
||||
}
|
||||
}()
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Service) requestProviderCancellation(ctx context.Context, task *model.Task) error {
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"math"
|
||||
"time"
|
||||
@@ -10,21 +11,26 @@ import (
|
||||
|
||||
const resourceDeletionLease = 2 * time.Minute
|
||||
|
||||
func (s *Service) startResourceDeletionWorker() {
|
||||
go func() {
|
||||
func (s *Service) startResourceDeletionWorker(ctx context.Context) {
|
||||
s.runWorkerLoop(func(ctx context.Context) {
|
||||
s.drainResourceDeletionJobs(32)
|
||||
s.cleanupStaleAnnouncementImageDrafts()
|
||||
ticker := time.NewTicker(15 * time.Second)
|
||||
defer ticker.Stop()
|
||||
lastDraftCleanup := time.Now()
|
||||
for range ticker.C {
|
||||
s.drainResourceDeletionJobs(32)
|
||||
if time.Since(lastDraftCleanup) >= time.Hour {
|
||||
s.cleanupStaleAnnouncementImageDrafts()
|
||||
lastDraftCleanup = time.Now()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
s.drainResourceDeletionJobs(32)
|
||||
if time.Since(lastDraftCleanup) >= time.Hour {
|
||||
s.cleanupStaleAnnouncementImageDrafts()
|
||||
lastDraftCleanup = time.Now()
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Service) drainResourceDeletionJobs(limit int) {
|
||||
|
||||
@@ -26,6 +26,7 @@ type Service struct {
|
||||
redeemBatchMu sync.Mutex
|
||||
storageMu sync.Mutex
|
||||
storageTestMu sync.Mutex
|
||||
workerRuntimeMu sync.Mutex
|
||||
activeStorageTests map[string]bool
|
||||
characterTaskMu sync.Mutex
|
||||
activeCancels map[string]context.CancelFunc
|
||||
@@ -50,6 +51,8 @@ type Service struct {
|
||||
routeCatalogVersion int64
|
||||
routeHealthMu sync.Mutex
|
||||
routeHealthBlocked map[string]time.Time
|
||||
workers *workerRuntime
|
||||
updateManager UpdateManager
|
||||
}
|
||||
|
||||
const taskWorkerConcurrency = 3
|
||||
@@ -124,9 +127,39 @@ func (s *Service) taskBilling() *taskBillingCoordinator {
|
||||
}
|
||||
|
||||
func (s *Service) StartWorker() {
|
||||
s.taskWorker().start()
|
||||
s.startResourceDeletionWorker()
|
||||
s.startSkillSyncWorker()
|
||||
runtime := s.backgroundWorkers()
|
||||
ctx, started := runtime.start()
|
||||
if !started {
|
||||
return
|
||||
}
|
||||
s.taskWorker().start(ctx)
|
||||
s.startResourceDeletionWorker(ctx)
|
||||
s.startSkillSyncWorker(ctx)
|
||||
}
|
||||
|
||||
func (s *Service) BeginDrain() { s.backgroundWorkers().beginDrain() }
|
||||
|
||||
func (s *Service) StopWorker(ctx context.Context) error { return s.backgroundWorkers().stop(ctx) }
|
||||
|
||||
func (s *Service) IsDraining() bool { return s.backgroundWorkers().isDraining() }
|
||||
|
||||
func (s *Service) ActiveWorkerTasks() int64 { return s.backgroundWorkers().activeTaskCount() }
|
||||
|
||||
func (s *Service) backgroundWorkers() *workerRuntime {
|
||||
s.workerRuntimeMu.Lock()
|
||||
defer s.workerRuntimeMu.Unlock()
|
||||
if s.workers == nil {
|
||||
s.workers = &workerRuntime{}
|
||||
}
|
||||
return s.workers
|
||||
}
|
||||
|
||||
func (s *Service) runWorkerLoop(fn func(context.Context)) bool {
|
||||
return s.backgroundWorkers().goLoop(fn)
|
||||
}
|
||||
|
||||
func (s *Service) runWorkerTask(fn func()) bool {
|
||||
return s.backgroundWorkers().goTask(fn)
|
||||
}
|
||||
|
||||
func (s *Service) CreateSession(userID string, req CreateSessionRequest) (*SessionDetail, error) {
|
||||
|
||||
@@ -208,23 +208,32 @@ func (s *Service) SyncGitHubSkill(userID string, skillID string) (*SkillItem, er
|
||||
return s.SkillDetail(userID, skill.ID)
|
||||
}
|
||||
|
||||
func (s *Service) startSkillSyncWorker() {
|
||||
go func() {
|
||||
func (s *Service) startSkillSyncWorker(ctx context.Context) {
|
||||
s.runWorkerLoop(func(ctx context.Context) {
|
||||
ticker := time.NewTicker(6 * time.Hour)
|
||||
defer ticker.Stop()
|
||||
for range ticker.C {
|
||||
skills, err := s.repo.AutoUpdatingGitHubSkills()
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
for index := range skills {
|
||||
_ = s.syncGitHubSkill(&skills[index])
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
skills, err := s.repo.AutoUpdatingGitHubSkills()
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
for index := range skills {
|
||||
_ = s.syncGitHubSkillWithContext(ctx, &skills[index])
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
})
|
||||
}
|
||||
|
||||
func (s *Service) syncGitHubSkill(skill *model.Skill) error {
|
||||
return s.syncGitHubSkillWithContext(context.Background(), skill)
|
||||
}
|
||||
|
||||
func (s *Service) syncGitHubSkillWithContext(ctx context.Context, skill *model.Skill) error {
|
||||
now := time.Now()
|
||||
spec, err := parseGitHubSkillURL(skill.SourceURL, skill.SourceRef, skill.SourceSubdir)
|
||||
if err != nil {
|
||||
@@ -234,7 +243,7 @@ func (s *Service) syncGitHubSkill(skill *model.Skill) error {
|
||||
_ = s.repo.SaveSkill(skill)
|
||||
return err
|
||||
}
|
||||
archive, commit, canonicalURL, resolvedRef, err := fetchGitHubSkillArchive(context.Background(), spec)
|
||||
archive, commit, canonicalURL, resolvedRef, err := fetchGitHubSkillArchive(ctx, spec)
|
||||
skill.LastCheckedAt = &now
|
||||
if err != nil {
|
||||
skill.SyncStatus = "failed"
|
||||
|
||||
@@ -13,6 +13,9 @@ import (
|
||||
// CreateTask 收敛任务进入系统前的 admission 流程：输入标准化、逻辑模型路由、
|
||||
// 能力/额度校验和持久化。执行阶段由 worker 与 provider 相关模块负责。
|
||||
func (s *Service) CreateTask(userID string, req CreateTaskRequest) (*model.Task, error) {
|
||||
if s.IsDraining() {
|
||||
return nil, &AppError{Status: 503, Code: 503, Message: "服务正在维护，暂不接受新的生成任务", Retryable: true}
|
||||
}
|
||||
prompt := strings.TrimSpace(req.Prompt)
|
||||
if prompt == "" {
|
||||
return nil, errors.New("prompt is required")
|
||||
|
||||
@@ -31,6 +31,9 @@ func (s *Service) taskLifecycle() *taskLifecycleCoordinator {
|
||||
|
||||
func (w *taskLifecycleCoordinator) retryTask(userID string, id string) (*model.Task, error) {
|
||||
s := w.service
|
||||
if s.IsDraining() {
|
||||
return nil, &AppError{Status: 503, Code: 503, Message: "服务正在维护，暂不接受任务重试", Retryable: true}
|
||||
}
|
||||
task, err := s.repo.TaskForUser(userID, id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -156,13 +159,22 @@ func (w *taskLifecycleCoordinator) cancelTask(_ context.Context, userID string,
|
||||
// 上游取消可能需要轮询确认，不能阻塞取消接口；请求上下文也不能因
|
||||
// 浏览器关闭而中断。后台对账会继续负责退款或费用核对。
|
||||
cancelTask := *task
|
||||
go func() {
|
||||
started := s.runWorkerTask(func() {
|
||||
requestCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
if err := s.requestProviderCancellation(requestCtx, &cancelTask); err != nil {
|
||||
_ = s.log(cancelTask.UserID, cancelTask.ID, "error", "发送上游取消请求失败", err.Error())
|
||||
}
|
||||
}()
|
||||
})
|
||||
if !started {
|
||||
// drain 后不能启动未登记 goroutine；当前 HTTP 请求同步完成首次取消，
|
||||
// http.Server.Shutdown 会等待该请求，后续确认仍由持久化对账恢复。
|
||||
requestCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
if err := s.requestProviderCancellation(requestCtx, &cancelTask); err != nil {
|
||||
_ = s.log(cancelTask.UserID, cancelTask.ID, "error", "发送上游取消请求失败", err.Error())
|
||||
}
|
||||
cancel()
|
||||
}
|
||||
} else {
|
||||
var billingErr error
|
||||
if originalStatus == model.TaskStatusQueued {
|
||||
|
||||
@@ -28,21 +28,24 @@ func (s *Service) taskWorker() *taskWorkerCoordinator {
|
||||
return newTaskWorkerCoordinator(s)
|
||||
}
|
||||
|
||||
func (w *taskWorkerCoordinator) start() {
|
||||
func (w *taskWorkerCoordinator) start(ctx context.Context) {
|
||||
s := w.service
|
||||
s.startTextReplayCleanup()
|
||||
s.startProviderCancellationReconciliation()
|
||||
s.startBillingReviewAudit()
|
||||
go func() {
|
||||
s.startTextReplayCleanup(ctx)
|
||||
s.startProviderCancellationReconciliation(ctx)
|
||||
s.startBillingReviewAudit(ctx)
|
||||
s.runWorkerLoop(func(ctx context.Context) {
|
||||
slots := make(chan struct{}, maxChannelConcurrencyLimit)
|
||||
dispatch := func() {
|
||||
if ctx.Err() != nil || s.IsDraining() {
|
||||
return
|
||||
}
|
||||
setting, err := s.runtimeConcurrencySetting()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
workerConcurrency := setting.WorkerConcurrency
|
||||
for len(slots) < workerConcurrency {
|
||||
releaseGlobal, acquired, err := s.coordinator.acquire(context.Background(), "workers", workerConcurrency, 45*time.Minute)
|
||||
releaseGlobal, acquired, err := s.coordinator.acquire(ctx, "workers", workerConcurrency, 45*time.Minute)
|
||||
if err != nil || !acquired {
|
||||
return
|
||||
}
|
||||
@@ -52,22 +55,33 @@ func (w *taskWorkerCoordinator) start() {
|
||||
return
|
||||
}
|
||||
slots <- struct{}{}
|
||||
go func(task *model.Task) {
|
||||
started := s.runWorkerTask(func() {
|
||||
defer func() { <-slots; releaseGlobal() }()
|
||||
if err := w.processClaimedTask(task); err != nil {
|
||||
_ = s.log(task.UserID, task.ID, "error", "后台任务处理失败", err.Error())
|
||||
}
|
||||
}(task)
|
||||
})
|
||||
if !started {
|
||||
<-slots
|
||||
releaseGlobal()
|
||||
_ = s.repo.ReleaseTaskLease(task.ID, s.workerID)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
dispatch()
|
||||
ticker := time.NewTicker(2 * time.Second)
|
||||
defer ticker.Stop()
|
||||
for range ticker.C {
|
||||
dispatch()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
dispatch()
|
||||
}
|
||||
}
|
||||
}()
|
||||
})
|
||||
}
|
||||
|
||||
func (w *taskWorkerCoordinator) processNextTask() error {
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"log"
|
||||
@@ -156,18 +157,23 @@ func taskResultText(raw string) string {
|
||||
return result.Text
|
||||
}
|
||||
|
||||
func (s *Service) startTextReplayCleanup() {
|
||||
func (s *Service) startTextReplayCleanup(ctx context.Context) {
|
||||
cleanup := func() {
|
||||
if _, err := s.CleanupTaskTextReplay(); err != nil {
|
||||
log.Printf("text replay cleanup failed: %v", err)
|
||||
}
|
||||
}
|
||||
cleanup()
|
||||
go func() {
|
||||
s.runWorkerLoop(func(ctx context.Context) {
|
||||
cleanup()
|
||||
ticker := time.NewTicker(time.Hour)
|
||||
defer ticker.Stop()
|
||||
for range ticker.C {
|
||||
cleanup()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
cleanup()
|
||||
}
|
||||
}
|
||||
}()
|
||||
})
|
||||
}
|
||||
|
||||
95
backend/internal/service/worker_runtime.go
Normal file
95
backend/internal/service/worker_runtime.go
Normal file
@@ -0,0 +1,95 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
)
|
||||
|
||||
type workerRuntime struct {
|
||||
mu sync.Mutex
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
wg sync.WaitGroup
|
||||
started bool
|
||||
draining atomic.Bool
|
||||
activeTasks atomic.Int64
|
||||
}
|
||||
|
||||
func (r *workerRuntime) start() (context.Context, bool) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
if r.started {
|
||||
return r.ctx, false
|
||||
}
|
||||
r.ctx, r.cancel = context.WithCancel(context.Background())
|
||||
r.started = true
|
||||
r.draining.Store(false)
|
||||
return r.ctx, true
|
||||
}
|
||||
|
||||
func (r *workerRuntime) goLoop(fn func(context.Context)) bool {
|
||||
r.mu.Lock()
|
||||
if !r.started || r.draining.Load() {
|
||||
r.mu.Unlock()
|
||||
return false
|
||||
}
|
||||
ctx := r.ctx
|
||||
r.wg.Add(1)
|
||||
r.mu.Unlock()
|
||||
go func() {
|
||||
defer r.wg.Done()
|
||||
fn(ctx)
|
||||
}()
|
||||
return true
|
||||
}
|
||||
|
||||
func (r *workerRuntime) goTask(fn func()) bool {
|
||||
r.mu.Lock()
|
||||
if !r.started || r.draining.Load() {
|
||||
r.mu.Unlock()
|
||||
return false
|
||||
}
|
||||
r.wg.Add(1)
|
||||
r.activeTasks.Add(1)
|
||||
r.mu.Unlock()
|
||||
go func() {
|
||||
defer r.wg.Done()
|
||||
defer r.activeTasks.Add(-1)
|
||||
fn()
|
||||
}()
|
||||
return true
|
||||
}
|
||||
|
||||
func (r *workerRuntime) beginDrain() {
|
||||
r.mu.Lock()
|
||||
r.draining.Store(true)
|
||||
if r.cancel != nil {
|
||||
r.cancel()
|
||||
}
|
||||
r.mu.Unlock()
|
||||
}
|
||||
|
||||
func (r *workerRuntime) stop(ctx context.Context) error {
|
||||
r.beginDrain()
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
r.wg.Wait()
|
||||
close(done)
|
||||
}()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-done:
|
||||
r.mu.Lock()
|
||||
r.started = false
|
||||
r.ctx = nil
|
||||
r.cancel = nil
|
||||
r.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
func (r *workerRuntime) isDraining() bool { return r.draining.Load() }
|
||||
|
||||
func (r *workerRuntime) activeTaskCount() int64 { return r.activeTasks.Load() }
|
||||
73
backend/internal/service/worker_runtime_test.go
Normal file
73
backend/internal/service/worker_runtime_test.go
Normal file
@@ -0,0 +1,73 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestWorkerRuntimeStopsLoopsAndWaitsForTasks(t *testing.T) {
|
||||
runtime := &workerRuntime{}
|
||||
ctx, started := runtime.start()
|
||||
if !started {
|
||||
t.Fatal("worker runtime should start")
|
||||
}
|
||||
loopStopped := make(chan struct{})
|
||||
if !runtime.goLoop(func(context.Context) {
|
||||
<-ctx.Done()
|
||||
close(loopStopped)
|
||||
}) {
|
||||
t.Fatal("worker loop should start")
|
||||
}
|
||||
taskRelease := make(chan struct{})
|
||||
if !runtime.goTask(func() { <-taskRelease }) {
|
||||
t.Fatal("worker task should start")
|
||||
}
|
||||
|
||||
stopCtx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
|
||||
defer cancel()
|
||||
if err := runtime.stop(stopCtx); !errors.Is(err, context.DeadlineExceeded) {
|
||||
t.Fatalf("stop should wait for the active task, got %v", err)
|
||||
}
|
||||
select {
|
||||
case <-loopStopped:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("worker loop did not observe drain cancellation")
|
||||
}
|
||||
if runtime.activeTaskCount() != 1 {
|
||||
t.Fatalf("active task count = %d, want 1", runtime.activeTaskCount())
|
||||
}
|
||||
|
||||
close(taskRelease)
|
||||
finalCtx, finalCancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer finalCancel()
|
||||
if err := runtime.stop(finalCtx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if runtime.activeTaskCount() != 0 {
|
||||
t.Fatalf("active task count = %d, want 0", runtime.activeTaskCount())
|
||||
}
|
||||
if runtime.goTask(func() {}) {
|
||||
t.Fatal("drained runtime should reject new tasks")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTaskWritesAreRejectedDuringDrain(t *testing.T) {
|
||||
runtime := &workerRuntime{}
|
||||
runtime.start()
|
||||
runtime.beginDrain()
|
||||
svc := &Service{workers: runtime}
|
||||
|
||||
if _, err := svc.CreateTask("user-1", CreateTaskRequest{}); !isAppErrorStatus(err, 503) {
|
||||
t.Fatalf("CreateTask during drain error = %v, want 503", err)
|
||||
}
|
||||
if _, err := svc.RetryTask("user-1", "task-1"); !isAppErrorStatus(err, 503) {
|
||||
t.Fatalf("RetryTask during drain error = %v, want 503", err)
|
||||
}
|
||||
}
|
||||
|
||||
func isAppErrorStatus(err error, status int) bool {
|
||||
var appErr *AppError
|
||||
return errors.As(err, &appErr) && appErr.Status == status
|
||||
}
|
||||
89
backend/internal/updaterclient/client.go
Normal file
89
backend/internal/updaterclient/client.go
Normal file
@@ -0,0 +1,89 @@
|
||||
package updaterclient
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"infinite-canvas/backend/internal/hostupdate"
|
||||
)
|
||||
|
||||
type Client struct {
|
||||
baseURL string
|
||||
token string
|
||||
http *http.Client
|
||||
}
|
||||
|
||||
func New(socketPath, token string) *Client {
|
||||
dialer := &net.Dialer{Timeout: 3 * time.Second}
|
||||
transport := &http.Transport{
|
||||
DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
|
||||
return dialer.DialContext(ctx, "unix", socketPath)
|
||||
},
|
||||
DisableKeepAlives: true,
|
||||
}
|
||||
return &Client{baseURL: "http://unix", token: strings.TrimSpace(token), http: &http.Client{Transport: transport, Timeout: 15 * time.Second}}
|
||||
}
|
||||
|
||||
func (c *Client) Status(ctx context.Context) (hostupdate.Status, error) {
|
||||
return c.request(ctx, http.MethodGet, "/v1/status", nil)
|
||||
}
|
||||
|
||||
func (c *Client) Check(ctx context.Context) (hostupdate.Status, error) {
|
||||
return c.request(ctx, http.MethodPost, "/v1/check", struct{}{})
|
||||
}
|
||||
|
||||
func (c *Client) Start(ctx context.Context, targetVersion string) (hostupdate.Status, error) {
|
||||
return c.request(ctx, http.MethodPost, "/v1/update", hostupdate.StartRequest{TargetVersion: targetVersion})
|
||||
}
|
||||
|
||||
func (c *Client) Rollback(ctx context.Context, reason string) (hostupdate.Status, error) {
|
||||
return c.request(ctx, http.MethodPost, "/v1/rollback", hostupdate.RollbackRequest{Reason: reason})
|
||||
}
|
||||
|
||||
func (c *Client) request(ctx context.Context, method, path string, payload any) (hostupdate.Status, error) {
|
||||
var body io.Reader
|
||||
if payload != nil {
|
||||
encoded, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return hostupdate.Status{}, err
|
||||
}
|
||||
body = bytes.NewReader(encoded)
|
||||
}
|
||||
request, err := http.NewRequestWithContext(ctx, method, c.baseURL+path, body)
|
||||
if err != nil {
|
||||
return hostupdate.Status{}, err
|
||||
}
|
||||
request.Header.Set("Authorization", "Bearer "+c.token)
|
||||
request.Header.Set("Accept", "application/json")
|
||||
if payload != nil {
|
||||
request.Header.Set("Content-Type", "application/json")
|
||||
}
|
||||
response, err := c.http.Do(request)
|
||||
if err != nil {
|
||||
return hostupdate.Status{}, fmt.Errorf("连接 Host Updater：%w", err)
|
||||
}
|
||||
defer response.Body.Close()
|
||||
limited := io.LimitReader(response.Body, 2<<20)
|
||||
if response.StatusCode < 200 || response.StatusCode >= 300 {
|
||||
var failure struct {
|
||||
Error string `json:"error"`
|
||||
Data hostupdate.Status `json:"data"`
|
||||
}
|
||||
if err := json.NewDecoder(limited).Decode(&failure); err == nil && failure.Error != "" {
|
||||
return failure.Data, fmt.Errorf("Host Updater：%s", failure.Error)
|
||||
}
|
||||
return hostupdate.Status{}, fmt.Errorf("Host Updater 返回 HTTP %d", response.StatusCode)
|
||||
}
|
||||
var status hostupdate.Status
|
||||
if err := json.NewDecoder(limited).Decode(&status); err != nil {
|
||||
return hostupdate.Status{}, fmt.Errorf("解析 Host Updater 响应：%w", err)
|
||||
}
|
||||
return status, nil
|
||||
}
|
||||
Reference in New Issue
Block a user