package localapi import ( "context" "encoding/json" "fmt" "os" "path/filepath" "sync" "time" ) // formatJob is the persisted record of the most recent / in-flight disk format (F20-BUG3). It exists so // that (a) an mkfs runs DETACHED from the HTTP request — a client/request deadline can no longer SIGKILL // an in-progress mkfs and corrupt the disk — and (b) an in-progress format SURVIVES an agent restart: // on boot, a record still in `running` is re-resolved by durable-id and re-run (mkfs is idempotent). type formatJob struct { JobID string `json:"job_id"` Device string `json:"device"` DurableID string `json:"durable_id"` // "" for a blank (benign) format — never auto-recovered FSType string `json:"fstype"` Phase string `json:"phase"` // running | done | failed Error string `json:"error,omitempty"` StartedAt string `json:"started_at"` UpdatedAt string `json:"updated_at"` } const ( formatPhaseRunning = "running" formatPhaseDone = "done" formatPhaseFailed = "failed" ) // FormatJobStore persists the single most-recent format job (atomic tmp+rename, 0600). One slot: a host // formats one device at a time. Mirrors storage.IntentStore. type FormatJobStore struct { path string mu sync.Mutex cur *formatJob } // OpenFormatJobStore loads (or initializes) the store. Missing file = no job; corrupt file = error. func OpenFormatJobStore(path string) (*FormatJobStore, error) { s := &FormatJobStore{path: path} data, err := os.ReadFile(path) if err != nil { if os.IsNotExist(err) { return s, nil } return nil, fmt.Errorf("format-job store: read %s: %w", path, err) } if len(data) > 0 { var j formatJob if err := json.Unmarshal(data, &j); err != nil { return nil, fmt.Errorf("format-job store: parse %s: %w", path, err) } s.cur = &j } return s, nil } func (s *FormatJobStore) get() *formatJob { s.mu.Lock() defer s.mu.Unlock() if s.cur == nil { return nil } cp := *s.cur return &cp } func (s *FormatJobStore) save(j *formatJob) error { s.mu.Lock() defer s.mu.Unlock() cp := *j s.cur = &cp data, err := json.MarshalIndent(s.cur, "", " ") if err != nil { return err } if err := os.MkdirAll(filepath.Dir(s.path), 0o755); err != nil { return err } tmp := s.path + ".tmp" if err := os.WriteFile(tmp, data, 0o600); err != nil { os.Remove(tmp) return err } return os.Rename(tmp, s.path) } // startFormatDetached persists a `running` record and runs mkfs in a goroutine off s.baseCtx (NOT the // request context), with a long bound. It returns a channel that yields the format error (nil on // success). The caller may stop waiting (client disconnect) without killing the mkfs — the goroutine // runs to completion and records the outcome. device is the ALREADY anti-retarget-resolved device; the // record carries durableID so a restart can re-resolve + re-run. func (s *Server) startFormatDetached(device, durableID, fstype string) <-chan error { base := s.baseCtx if base == nil { base = context.Background() } job := &formatJob{ JobID: s.nowFn().UTC().Format("20060102T150405Z"), Device: device, DurableID: durableID, FSType: fstype, Phase: formatPhaseRunning, StartedAt: s.nowFn().UTC().Format(time.RFC3339), UpdatedAt: s.nowFn().UTC().Format(time.RFC3339), } if s.formatJobs != nil { if err := s.formatJobs.save(job); err != nil { s.logger.Warn("format-job: could not persist running record (format will still run)", "device", device, "err", err) } } done := make(chan error, 1) go func() { ctx, cancel := context.WithTimeout(base, 60*time.Minute) defer cancel() err := s.disks.Format(ctx, device, fstype) s.finishFormatJob(job, err) done <- err }() return done } // finishFormatJob updates the persisted record to done/failed. func (s *Server) finishFormatJob(job *formatJob, err error) { if s.formatJobs == nil { return } job.UpdatedAt = s.nowFn().UTC().Format(time.RFC3339) if err != nil { job.Phase = formatPhaseFailed job.Error = err.Error() } else { job.Phase = formatPhaseDone } if serr := s.formatJobs.save(job); serr != nil { s.logger.Warn("format-job: could not persist final record", "device", job.Device, "phase", job.Phase, "err", serr) } } // RecoverFormatJob, on agent startup, completes a format that an agent restart interrupted (F20-BUG3 + // the operator's "survive a restart" decision). A record still in `running` is re-resolved by its // durable-id (anti-retarget — a swapped/absent disk is NOT re-formatted) and the mkfs is re-run detached // (mkfs is idempotent). A blank (no-durable-id) format is NOT auto-re-run — it is marked failed for the // caller to retry, so recovery never formats a mutable /dev path. func (s *Server) RecoverFormatJob(ctx context.Context) { if s.formatJobs == nil { return } job := s.formatJobs.get() if job == nil || job.Phase != formatPhaseRunning { return } if job.DurableID == "" { s.logger.Warn("format-job recover: interrupted blank format — marking failed (retry needed; not auto-re-running a path-bound format)", "device", job.Device) s.finishFormatJob(job, fmt.Errorf("interrupted by agent restart; retry the format")) return } device, err := s.reresolveWipe(ctx, job.DurableID) if err != nil { s.logger.Warn("format-job recover: durable-id no longer resolves — NOT re-formatting (anti-retarget)", "durable_id", job.DurableID, "err", err) s.finishFormatJob(job, fmt.Errorf("durable-id %s did not re-resolve after restart: %w", job.DurableID, err)) return } s.logger.Warn("format-job recover: re-running interrupted format detached", "durable_id", job.DurableID, "device", device, "fstype", job.FSType) _ = s.startFormatDetached(device, job.DurableID, job.FSType) // detached; updates the record on completion } // nowFn returns the server clock (testable), defaulting to time.Now. func (s *Server) nowFn() time.Time { if s.now != nil { return s.now() } return time.Now() }