package api import ( "context" "database/sql" "encoding/json" "errors" "fmt" "git.misaka.ren/m1saka/dsh-web/internal/docker" "net/http" "net/url" "os" "path/filepath" "runtime" "syscall" "time" ) const imageRecipeRev = "7" var imagePlatform = runtime.GOOS + "/" + runtime.GOARCH const minimumBuildDiskBytes = uint64(2 << 30) var buildDiskAvailable = func(path string) (uint64, error) { var stat syscall.Statfs_t if err := syscall.Statfs(path, &stat); err != nil { return 0, err } return uint64(stat.Bavail) * uint64(stat.Bsize), nil } type inspectedImage struct { ID string `json:"Id"` Os, Architecture, Variant, Created string RepoTags []string Config struct{ Labels map[string]string } } func (im inspectedImage) platform() string { if im.Os == "" || im.Architecture == "" { return "" } p := im.Os + "/" + im.Architecture // Docker can report the default ARM64 variant explicitly. Keep its // identity consistent with build records and admission keys; preserve // all other variants so incompatible images remain rejected. if im.Architecture == "arm64" && im.Variant == "v8" { return p } if im.Variant != "" { p += "/" + im.Variant } return p } func (a *API) inspectLocalImage(ctx context.Context, id string) (*inspectedImage, error) { if id == "" { return nil, nil } var im inspectedImage if err := a.docker.Do(ctx, http.MethodGet, "/images/"+url.PathEscape(id)+"/json", nil, &im); err != nil { var apiErr *docker.APIError if errors.As(err, &apiErr) && apiErr.StatusCode == http.StatusNotFound { return nil, nil } return nil, err } return &im, nil } func (a *API) imageMatches(ctx context.Context, id, version string) (bool, error) { im, err := a.inspectLocalImage(ctx, id) if err != nil || im == nil { return false, err } return im.ID == id && im.platform() == imagePlatform && im.Config.Labels["dsh-mgr.role"] == "runtime" && im.Config.Labels["dsh-mgr.version"] == version && im.Config.Labels["dsh-mgr.recipe"] == imageRecipeRev, nil } var errImageIndexChanged = errors.New("image inventory changed during reconciliation") func (a *API) imageIndex(ctx context.Context) ([]imageRecord, error) { // Retry stale observations without holding the queue across Docker I/O. // Bound retries so sustained deletions cannot keep a request alive forever. for attempt := 0; attempt < 3; attempt++ { result, err := a.imageIndexSnapshot(ctx) if !errors.Is(err, errImageIndexChanged) { return result, err } } return nil, errImageIndexChanged } func (a *API) imageIndexSnapshot(ctx context.Context) ([]imageRecord, error) { // Phase 1: read the managed index under the queue lock (DB only). buildQueue.Lock() generation := a.imageDeleteGeneration err := a.ensureBuilds() if err == nil { _, err = a.s.DB.ExecContext(ctx, `INSERT OR IGNORE INTO images(image_id,version,recipe_rev,platform,tag,status,built_at) SELECT image_id,version,recipe_rev,platform,'','missing',COALESCE(finished_at,created_at) FROM image_builds WHERE status='success' AND image_id<>'' ORDER BY created_at DESC,id DESC`) } var rows *sql.Rows if err == nil { rows, err = a.s.DB.QueryContext(ctx, `SELECT image_id,version,COALESCE(recipe_rev,''),COALESCE(platform,''),tag,status,built_at, (SELECT count(*) FROM instances WHERE instances.image_id=images.image_id) FROM images ORDER BY built_at DESC,image_id`) } result := []imageRecord{} if err == nil { for rows.Next() { var im imageRecord if err = rows.Scan(&im.ImageID, &im.Version, &im.RecipeRev, &im.Platform, &im.Tag, &im.Status, &im.Created, &im.UsageCount); err != nil { break } im.ID = im.ImageID result = append(result, im) } scanErr := rows.Err() rows.Close() if err == nil { err = scanErr } } buildQueue.Unlock() if err != nil { return nil, err } // Phase 2: inspect Docker without the queue lock; per-image state comes // from real daemon observations, and instance creation must not wait here. known := make(map[string]bool, len(result)) for _, im := range result { known[im.ImageID] = true } var local []struct { ID string Labels map[string]string } { filters, _ := json.Marshal(map[string][]string{"label": {"dsh-mgr.role=runtime"}}) if err := a.docker.Do(ctx, http.MethodGet, "/images/json?filters="+url.QueryEscape(string(filters)), nil, &local); err != nil { return nil, err } } for _, im := range local { if im.ID != "" && !known[im.ID] && im.Labels["dsh-mgr.role"] == "runtime" { known[im.ID] = true result = append(result, imageRecord{ID: im.ID, ImageID: im.ID}) } } for i := range result { im := &result[i] im.Status = "missing" im.Admission = "existing_only" im.AdmissionReason = "local image missing" actual, err := a.inspectLocalImage(ctx, im.ImageID) if err != nil { return nil, err } if actual == nil { continue } im.Status = "unverified" im.Platform = actual.platform() im.Version = actual.Config.Labels["dsh-mgr.version"] im.RecipeRev = actual.Config.Labels["dsh-mgr.recipe"] im.Tag = "" if len(actual.RepoTags) > 0 { im.Tag = actual.RepoTags[0] } if actual.Created != "" { im.Created = actual.Created } if err := a.s.DB.QueryRowContext(ctx, "SELECT count(*) FROM instances WHERE image_id=?", im.ImageID).Scan(&im.UsageCount); err != nil { return nil, err } policy, err := a.evaluateImage(ctx, im.ImageID) if err != nil { return nil, err } im.Status, im.Admission, im.AdmissionReason, im.Selectable = policy.Status, policy.Admission, policy.AdmissionReason, policy.Selectable } // Phase 3: write observed state back under the queue lock (DB only). buildQueue.Lock() defer buildQueue.Unlock() if generation != a.imageDeleteGeneration { return nil, errImageIndexChanged } tx, err := a.s.DB.BeginTx(ctx, nil) if err != nil { return nil, err } defer tx.Rollback() present := make([]imageRecord, 0, len(result)) for _, im := range result { if im.Status == "missing" { if _, err = tx.ExecContext(ctx, "UPDATE image_builds SET status='missing',error='image no longer available or template identity is unknown' WHERE image_id=? AND status='success'", im.ImageID); err != nil { return nil, err } if _, err = tx.ExecContext(ctx, "DELETE FROM images WHERE image_id=?", im.ImageID); err != nil { return nil, err } continue } if _, err = tx.ExecContext(ctx, `INSERT INTO images(image_id,version,recipe_rev,platform,tag,status,built_at,usage_count) VALUES(?,?,?,?,?,?,?,?) ON CONFLICT(image_id) DO UPDATE SET version=excluded.version,recipe_rev=excluded.recipe_rev,platform=excluded.platform,tag=excluded.tag,status=excluded.status,built_at=excluded.built_at,usage_count=excluded.usage_count`, im.ImageID, im.Version, im.RecipeRev, im.Platform, im.Tag, im.Status, im.Created, im.UsageCount); err != nil { return nil, err } present = append(present, im) } if err = tx.Commit(); err != nil { return nil, err } return present, nil } func (a *API) StartImageBuilds() error { buildQueue.Lock() defer buildQueue.Unlock() if err := a.ensureBuilds(); err != nil { return err } a.startBuildWorkerLocked() return nil } func (a *API) checkBuildDisk(ctx context.Context) error { var info struct{ DockerRootDir string } if err := a.docker.Do(ctx, http.MethodGet, "/info", nil, &info); err != nil { return fmt.Errorf("build disk check: %w", err) } // A matching pathname alone cannot establish daemon filesystem identity. confirmed := os.Getenv("DSH_DOCKER_ROOT_VISIBLE") if !filepath.IsAbs(info.DockerRootDir) || confirmed != info.DockerRootDir { return errors.New("Docker disk capacity unknown: mount the daemon DockerRootDir at the identical path and set DSH_DOCKER_ROOT_VISIBLE to that path; new builds are blocked") } available, err := buildDiskAvailable(confirmed) if err != nil { return fmt.Errorf("Docker disk capacity unavailable: %w", err) } return requireBuildDisk(available) } func requireBuildDisk(available uint64) error { if available < minimumBuildDiskBytes { return fmt.Errorf("insufficient Docker disk space: %d available bytes, require %d", available, minimumBuildDiskBytes) } return nil } func (a *API) insertBuildRequest(b buildRecord, user string) error { tx, err := a.s.DB.Begin() if err != nil { return err } defer tx.Rollback() if _, err = tx.Exec("INSERT INTO image_builds(id,version,status,created_at,requested_by,recipe_rev,platform,experimental,release_id,source_repository,source_commit) VALUES(?,?,?,?,?,?,?,?,?,?,?)", b.ID, b.Version, b.Status, b.Created, b.RequestedBy, b.RecipeRev, b.Platform, b.Experimental, b.ReleaseID, b.SourceRepository, b.SourceCommit); err != nil { return err } if _, err = tx.Exec("INSERT INTO image_build_requests(user_id,build_id,created_at) VALUES(?,?,?)", user, b.ID, b.Created); err != nil { return err } return tx.Commit() } func (a *API) reconcileBuilds(parent context.Context, records []buildRecord) error { ctx, cancel := context.WithTimeout(parent, 15*time.Second) defer cancel() for i := range records { b := &records[i] if b.Status != "success" { continue } actual, err := a.inspectLocalImage(ctx, b.ImageID) if err != nil { return err } if actual == nil || actual.ID != b.ImageID || actual.Config.Labels["dsh-mgr.role"] != "runtime" || actual.Config.Labels["dsh-mgr.version"] != b.Version || actual.Config.Labels["dsh-mgr.recipe"] != b.RecipeRev || actual.platform() != b.Platform || !releaseLabelsMatch(actual, *b) { b.Status, b.Error = "missing", "image no longer available" if _, err := a.s.DB.Exec("UPDATE image_builds SET status=?,error=? WHERE id=? AND status='success'", b.Status, b.Error, b.ID); err != nil { return err } } } return nil }