felhom-controller/controller/internal/web/storage_init_job.go

package web

import (
	"context"
	"sync"
	"time"

	"gitea.dooplex.hu/admin/felhom-controller/internal/logx"
)

// Drive-initialize orchestration (F6, VALIDATION-n100). POST /api/storage/init no longer runs the
// format→mount→register chain ON THE REQUEST CONTEXT — a closed tab or a lost connection cancelled
// that context after mkfs (the agent's mkfs continues detached, but the controller's mount+register
// leg was aborted, leaving a formatted-but-unusable drive). It now starts a DETACHED job (the
// netstorage-add shape: single-flight, deep-copied status, register-LAST) the wizard polls via
// GET /api/storage/init/status. The chain reuses runStorageInit unchanged (format via the agent's
// crash-safe detached mkfs → resolve UUID → AssignDisk → attachIntoGuest → registerStoragePath →
// SyncFileBrowserMounts); this file only moves it off the request context and makes it pollable.
//
// Crash-safety (Scenario B): registerStoragePath is the LAST step (marker-last), and every prior
// step is idempotent (mkfs idempotent; AssignDisk idempotent; AddStoragePath dedups a re-register).
// So the worst outcome of a controller crash mid-chain is a formatted+mounted-but-unregistered
// drive surfaced by the drive list — never a registered-but-broken path, and never a duplicate
// registry entry: a re-run of the wizard self-heals to the terminal state.

// storageInitJob is the poll-visible orchestration state (GET /api/storage/init/status).
type storageInitJob struct {
	Device    string    `json:"device"`
	Where     string    `json:"where,omitempty"` // the registered stable path (done only)
	Phase     string    `json:"phase"`           // formatting | mounting | registering | done | failed | needs_confirmation | refused
	Error     string    `json:"error,omitempty"`
	Reason    string    `json:"reason,omitempty"`     // refusal/confirm reason (Hungarian, from the agent)
	DurableID string    `json:"durable_id,omitempty"` // needs_confirmation: the durable id to confirm against
	Opsign    string    `json:"opsign,omitempty"`     // refused: the operator opsign command
	StartedAt time.Time `json:"started_at"`
	UpdatedAt time.Time `json:"updated_at"`
}

const (
	storageInitPhaseFormatting   = "formatting"
	storageInitPhaseMounting     = "mounting"
	storageInitPhaseRegistering  = "registering"
	storageInitPhaseDone         = "done"
	storageInitPhaseFailed       = "failed"
	storageInitPhaseNeedsConfirm = "needs_confirmation"
	storageInitPhaseRefused      = "refused"
)

// storageInitDeadline bounds the WHOLE chain on the detached context. The dominant term is the
// agent's mkfs (bounded to 60 min agent-side); a wall-clock ceiling well above any real format
// keeps a wedged agent from pinning the single-flight slot forever.
const storageInitDeadline = 65 * time.Minute

// storageInitParams carries the validated init request into the detached job.
type storageInitParams struct {
	device, fstype, where, label, durableID string
	setDefault, confirmed                   bool
}

// storageInitState is the single-flight slot (netAddState shape: acquire/release/set/snapshot,
// deep-copied status). One guest initializes one drive at a time.
type storageInitState struct {
	mu      sync.Mutex
	running bool
	cur     *storageInitJob
}

func (s *storageInitState) acquire(job *storageInitJob) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.running {
		return false
	}
	s.running = true
	cp := *job
	s.cur = &cp
	return true
}

func (s *storageInitState) release() {
	s.mu.Lock()
	s.running = false
	s.mu.Unlock()
}

func (s *storageInitState) set(job *storageInitJob) {
	s.mu.Lock()
	cp := *job
	s.cur = &cp
	s.mu.Unlock()
}

// snapshot returns a copy of the last / in-flight job (nil = never ran).
func (s *storageInitState) snapshot() *storageInitJob {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.cur == nil {
		return nil
	}
	cp := *s.cur
	return &cp
}

// startStorageInit claims the single-flight slot and launches the detached init chain. false = an
// init is already in flight (Scenario/edge: never two chains on one device).
func (s *Server) startStorageInit(agent diskAgent, p storageInitParams) bool {
	now := time.Now().UTC()
	job := &storageInitJob{
		Device: p.device, Where: p.where, Phase: storageInitPhaseFormatting,
		StartedAt: now, UpdatedAt: now,
	}
	if !s.storageInit.acquire(job) {
		return false
	}
	go s.runStorageInitJob(agent, p, job)
	return true
}

// runStorageInitJob drives runStorageInit on a DETACHED context (a closed tab can no longer abort
// the mount+register leg — F6) and records the phase/outcome in the poll slot.
func (s *Server) runStorageInitJob(agent diskAgent, p storageInitParams, job *storageInitJob) {
	defer s.storageInit.release()
	ctx, cancel := context.WithTimeout(context.Background(), storageInitDeadline)
	defer cancel()
	start := time.Now()
	logx.Infof(s.logger, "[web] storage init %q started (device %s, fs %s)", p.where, p.device, p.fstype)

	progress := func(phase string) {
		logx.Debugf(s.logger, "[web] storage init %q phase %s -> %s (%dms elapsed)",
			p.where, job.Phase, phase, time.Since(start).Milliseconds())
		job.Phase = phase
		job.UpdatedAt = time.Now().UTC()
		s.storageInit.set(job)
	}

	res, err := s.runStorageInit(ctx, agent, p.device, p.fstype, p.where, p.label, p.setDefault, p.confirmed, p.durableID, progress)
	job.UpdatedAt = time.Now().UTC()
	switch {
	case err != nil:
		job.Phase = storageInitPhaseFailed
		job.Error = err.Error()
		logx.Warnf(s.logger, "[web] storage init %q failed: %v (%dms)", p.where, err, time.Since(start).Milliseconds())
	case res.NeedsConfirmation:
		job.Phase = storageInitPhaseNeedsConfirm
		job.DurableID = res.DurableID
		job.Reason = res.Reason
	case res.Refused:
		job.Phase = storageInitPhaseRefused
		job.Reason = res.Reason
		job.Opsign = res.Opsign
	default:
		job.Phase = storageInitPhaseDone
		job.Where = res.Where
		logx.Infof(s.logger, "[web] storage init done: %s registered (%dms)", res.Where, time.Since(start).Milliseconds())
	}
	s.storageInit.set(job)
}