1package oshost23import (4 "fmt"5 "io"6 "log/slog"7 "os"8 "sort"9 "sync"10 "sync/atomic"11 "time"1213 "github.com/FelineStateMachine/atlas/format/bundle"14 "github.com/FelineStateMachine/atlas/internal/app/hostenv"15 "github.com/FelineStateMachine/atlas/internal/logging"16)1718// closeGrace is how long a retired bundle stays open after a rescan replaces19// it. A request served out of the previous snapshot finishes in milliseconds;20// the grace period outlives any of them without keeping descriptors around.21const closeGrace = 5 * time.Second2223// Volumes is a library kept as a directory of .atlas files.24//25// It scans at construction and on import, and never watches (issue #5 §2,26// decision 15): a file dropped in from outside appears at the next launch,27// which is what a reader expects of "open Atlas and see my registry". The28// fold that decides which build serves is format/bundle's and is pure; this29// type is the walk, the open files, and the swap.30//31// Every rescan builds a whole snapshot and swaps it in at once, so a request32// sees one consistent library however the directory is churning.33type Volumes struct {34 dir string35 snap atomic.Pointer[snapshot]3637 // scanning runs alone: two concurrent scans of one directory would race38 // to swap snapshots and could close each other's bundles.39 scanning sync.Mutex40}4142// snapshot is the library at one moment: the winning build per slug, the open43// readers behind them, and the order the handler lists them in.44type snapshot struct {45 serving map[string]bundle.Descriptor46 open map[string]*volume47 order []string48}4950// volume is one serving build with its archive held open, so answering for an51// entry is a seek and a read rather than a fresh open of a file that may be52// hundreds of megabytes.53type volume struct{ reader *bundle.Reader }5455func (v *volume) Manifest() bundle.Manifest { return v.reader.Manifest }5657func (v *volume) Open(entry string) (io.ReadCloser, int64, error) {58 return v.reader.OpenEntry(entry)59}6061// NewVolumes answers for dir, which need not exist yet: a directory that is62// missing or empty is a library with no volumes, not an error. The first scan63// happens here, so a caller that gets a store gets a library.64func NewVolumes(dir string) (*Volumes, error) {65 v := &Volumes{dir: dir}66 v.snap.Store(&snapshot{serving: map[string]bundle.Descriptor{}, open: map[string]*volume{}})67 if _, err := v.Rescan(); err != nil {68 return nil, err69 }70 return v, nil71}7273// Location is the directory the library is kept in.74func (v *Volumes) Location() string { return v.dir }7576// Volumes lists the serving build of every installed volume, sorted by slug.77func (v *Volumes) Volumes() []hostenv.Volume {78 held := v.snap.Load()79 out := make([]hostenv.Volume, 0, len(held.order))80 for _, slug := range held.order {81 out = append(out, held.open[slug])82 }83 return out84}8586// Close closes every reader the current snapshot holds and leaves the store87// empty. The desktop host never calls it — a process done with its library is88// a process that is exiting — but a caller that opened a library it must89// later delete has to let the files go first, which on Windows is a rule90// rather than a courtesy.91func (v *Volumes) Close() error {92 v.scanning.Lock()93 defer v.scanning.Unlock()94 previous := v.snap.Load()95 v.snap.Store(&snapshot{serving: map[string]bundle.Descriptor{}, open: map[string]*volume{}})96 var first error97 for _, held := range previous.open {98 if err := held.reader.Close(); err != nil && first == nil {99 first = err100 }101 }102 return first103}104105// Rescan reads the directory whole and swaps in what it finds, reporting the106// volumes whose serving build moved. A bundle that will not open is logged107// and passed over rather than failing the scan: one bad download should not108// take a library down with it.109func (v *Volumes) Rescan() ([]string, error) {110 v.scanning.Lock()111 defer v.scanning.Unlock()112113 descriptors, skipped, err := bundle.Scan(v.dir)114 if err != nil {115 return nil, fmt.Errorf("scan library: %w", err)116 }117 for _, passed := range skipped {118 slog.Warn("bundle skipped", logging.Op("scan"),119 logging.Path(passed.Locator), slog.Any("error", passed.Err))120 }121122 previous := v.snap.Load()123 winners := bundle.Fold(descriptors)124 next := &snapshot{125 serving: make(map[string]bundle.Descriptor, len(winners)),126 open: make(map[string]*volume, len(winners)),127 order: make([]string, 0, len(winners)),128 }129 for slug, winner := range winners {130 opened, err := carryOrOpen(previous, winner)131 if err != nil {132 slog.Warn("volume skipped", logging.Op("scan"),133 logging.Volume(slug), logging.Path(winner.Locator), slog.Any("error", err))134 continue135 }136 next.serving[slug] = winner137 next.open[slug] = opened138 next.order = append(next.order, slug)139 }140 sort.Strings(next.order)141 v.snap.Store(next)142143 retire(previous, next)144 changed := bundle.Changed(previous.serving, next.serving)145 for _, slug := range changed {146 if winner, still := next.serving[slug]; still {147 slog.Info("volume serving", logging.Op("scan"),148 logging.Volume(slug), logging.Stamp(bundle.ShortStamp(winner.Stamp)))149 } else {150 slog.Info("volume gone", logging.Op("scan"), logging.Volume(slug))151 }152 }153 return changed, nil154}155156// Install copies bundle content into the library under its versioned name and157// rescans. The content is staged to a temporary file first because the format158// validates a bundle before letting it in, and validation wants a file it can159// seek; the stage is removed either way.160func (v *Volumes) Install(name string, content io.Reader) (hostenv.Installed, error) {161 if err := os.MkdirAll(v.dir, 0o755); err != nil {162 return hostenv.Installed{}, fmt.Errorf("open library: %w", err)163 }164 // Staged inside the library so the install below is a rename within one165 // filesystem, and under a name that does not end in .atlas so a scan166 // running beside it passes the half-written file over.167 staged, err := os.CreateTemp(v.dir, ".importing-*")168 if err != nil {169 return hostenv.Installed{}, fmt.Errorf("stage %s: %w", name, err)170 }171 stage := staged.Name()172 defer os.Remove(stage)173 if _, err := io.Copy(staged, content); err != nil {174 staged.Close()175 return hostenv.Installed{}, fmt.Errorf("stage %s: %w", name, err)176 }177 if err := staged.Close(); err != nil {178 return hostenv.Installed{}, fmt.Errorf("stage %s: %w", name, err)179 }180181 standing := v.snap.Load().serving182 descriptor, err := bundle.Install(v.dir, stage)183 if err != nil {184 return hostenv.Installed{}, fmt.Errorf("%s: %w", name, err)185 }186 installed := hostenv.Installed{187 Slug: descriptor.Slug,188 Title: descriptor.Title,189 Stamp: descriptor.Stamp,190 Bytes: descriptor.Size,191 Already: standing[descriptor.Slug].Stamp == descriptor.Stamp,192 }193 slog.Info("bundle installed", logging.Op("install"),194 logging.Volume(descriptor.Slug), logging.Stamp(bundle.ShortStamp(descriptor.Stamp)),195 logging.Path(descriptor.Locator))196197 changed, err := v.Rescan()198 if err != nil {199 return installed, err200 }201 installed.Changed = changed202 return installed, nil203}204205// carryOrOpen reuses the reader already open for a build whose file has not206// moved, and opens the file fresh otherwise.207func carryOrOpen(previous *snapshot, winner bundle.Descriptor) (*volume, error) {208 if held, ok := previous.open[winner.Slug]; ok && held.reader.Path == winner.Locator {209 if info, err := os.Stat(winner.Locator); err == nil &&210 held.reader.Unchanged(info.Size(), info.ModTime()) {211 return held, nil212 }213 }214 reader, err := bundle.Open(winner.Locator)215 if err != nil {216 return nil, err217 }218 return &volume{reader: reader}, nil219}220221// retire closes the readers the previous snapshot held that the new one did222// not carry across, after a grace period long enough for any request already223// streaming out of one to finish.224func retire(previous, next *snapshot) {225 var closing []*volume226 for slug, held := range previous.open {227 if carried, still := next.open[slug]; still && carried == held {228 continue229 }230 closing = append(closing, held)231 }232 if len(closing) == 0 {233 return234 }235 time.AfterFunc(closeGrace, func() {236 for _, old := range closing {237 if err := old.reader.Close(); err != nil {238 slog.Warn("closing a retired bundle", logging.Op("scan"),239 logging.Path(old.reader.Path), slog.Any("error", err))240 }241 }242 })243}