package state import ( "encoding/json" "errors" "fmt" "os" "path/filepath" "time" ) type Phase string const ( PhaseDiscovered Phase = "discovered" PhaseReviewed Phase = "reviewed" PhaseArchived Phase = "archived" PhaseFailed Phase = "failed" // PhaseNeedsAttention is an album an unattended run left alone because // something about it needs a person: an unknown artist, a name that cannot // be normalised, a file that is not lossless. It is not a failure -- nothing // was written -- and the album is still sitting in staging. PhaseNeedsAttention Phase = "needs_attention" ) type Record struct { Source string `json:"source"` AlbumArtist string `json:"album_artist,omitempty"` Album string `json:"album,omitempty"` Phase Phase `json:"phase"` ArchiveDir string `json:"archive_dir,omitempty"` LibraryDir string `json:"library_dir,omitempty"` Tracks int `json:"tracks"` // Adopted marks a record built from an archive that was already on disk // rather than from a run that filed the album. Adopted bool `json:"adopted,omitempty"` Placeholder bool `json:"placeholder_cover"` LibraryBuilt bool `json:"library_built"` Issues []string `json:"issues,omitempty"` UpdatedAt time.Time `json:"updated_at"` } type Journal struct { Version int `json:"version"` Albums map[string]Record `json:"albums"` path string } func Open(path string) (*Journal, error) { j := &Journal{Version: 1, Albums: map[string]Record{}, path: path} data, err := os.ReadFile(path) if err != nil { if errors.Is(err, os.ErrNotExist) { return j, nil } return nil, err } if len(data) == 0 { return j, nil } if err := json.Unmarshal(data, j); err != nil { return nil, err } if j.Albums == nil { j.Albums = map[string]Record{} } j.path = path return j, nil } func (j *Journal) Path() string { return j.path } func (j *Journal) Get(source string) (Record, bool) { r, ok := j.Albums[source] return r, ok } func (j *Journal) Put(r Record) { r.UpdatedAt = time.Now().UTC() j.Albums[r.Source] = r } func (j *Journal) Delete(source string) { delete(j.Albums, source) } func (j *Journal) Save() error { if j.path == "" { return nil } if err := os.MkdirAll(filepath.Dir(j.path), 0o755); err != nil { return err } data, err := json.MarshalIndent(j, "", " ") if err != nil { return err } // A temp name of its own, so that two writers cannot rename each other's // half-written file into place. The fixed ".tmp" name meant the loser of a // race got "no such file or directory" from its own rename and lost the // record it was trying to write. tmp, err := os.CreateTemp(filepath.Dir(j.path), filepath.Base(j.path)+".tmp-*") if err != nil { return err } name := tmp.Name() if _, err := tmp.Write(data); err != nil { tmp.Close() os.Remove(name) return err } if err := tmp.Sync(); err != nil { tmp.Close() os.Remove(name) return err } if err := tmp.Close(); err != nil { os.Remove(name) return err } if err := os.Rename(name, j.path); err != nil { os.Remove(name) return err } return nil } // Update runs fn against the journal with the journal locked, and saves the // result when fn asks for that. The lock is what makes a read, a change and a // write one indivisible step. // // Every update used to be an unlocked open-modify-save. Two intake processes on // one root both read the journal, each filed its own album, and the second save // wrote back a snapshot taken before the first one, so the first album's record // vanished while the album itself stayed on disk. Nothing downstream could tell: // verify and audit work from the journal, so the album stopped existing as far as // intake was concerned, and the next resume filed it a second time. func Update(path string, fn func(*Journal) (bool, error)) error { if path == "" { return nil } if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { return err } unlock, err := lockJournal(path) if err != nil { return err } defer unlock() j, err := Open(path) if err != nil { return err } save, err := fn(j) if err != nil { return err } if !save { return nil } return j.Save() } const ( lockPoll = 25 * time.Millisecond lockWait = 15 * time.Second lockStale = 5 * time.Minute ) // lockJournal takes a cooperative lock beside the journal. It is a lock file // created with O_EXCL rather than flock, because flock is not available // everywhere intake builds and the window it protects is a few milliseconds of // file writing rather than a long-running operation. // // A lock older than lockStale is broken rather than waited on for ever: the // alternative is that a run killed while holding one stops every later run. func lockJournal(path string) (func(), error) { lock := path + ".lock" deadline := time.Now().Add(lockWait) for { f, err := os.OpenFile(lock, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600) if err == nil { fmt.Fprintf(f, "%d\n", os.Getpid()) f.Close() return func() { os.Remove(lock) }, nil } if !os.IsExist(err) { return nil, err } if fi, statErr := os.Stat(lock); statErr == nil && time.Since(fi.ModTime()) > lockStale { os.Remove(lock) continue } if time.Now().After(deadline) { return nil, fmt.Errorf("the journal at %s is locked by another intake run (%s); wait for it to finish, or delete that file if no run is active", path, lock) } time.Sleep(lockPoll) } }