| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "io/fs" |
| 9 | "log" |
| 10 | "os" |
| 11 | "path/filepath" |
| 12 | "strings" |
| 13 | |
| 14 | "github.com/carlosframework/platform/internal/moves" |
| 15 | "github.com/carlosframework/platform/internal/provision" |
| 16 | "github.com/carlosframework/platform/internal/release" |
| 17 | ) |
| 18 | |
| 19 | // The move log's outbox (Plan 6 Task 3, fix round 2, ruling 1). An entry |
| 20 | // whose write fails is kept on the box's disk, under |
| 21 | // <DataDir>/.moves/.log-outbox/<host>/<op>/<identity>.json, with the entry's |
| 22 | // exact bytes and the app it belongs to. Every pass retries every kept entry |
| 23 | // (flushLogOutbox), whatever its move has done since, and deletes the file |
| 24 | // only once the entry is created or found there. The directory's leading dot |
| 25 | // keeps it apart from the moves journal's host directories. |
| 26 | |
| 27 | // moveLogOutboxDir is the outbox's root. |
| 28 | func moveLogOutboxDir(dataDir string) string { |
| 29 | return filepath.Join(dataDir, ".moves", ".log-outbox") |
| 30 | } |
| 31 | |
| 32 | // moveLogPending is one kept entry: Entry is the bytes the log object holds |
| 33 | // (json.Marshal of the moves.LogEvent, as moves.WriteLog writes it). |
| 34 | type moveLogPending struct { |
| 35 | V int `json:"v"` |
| 36 | Account string `json:"account"` |
| 37 | App string `json:"app"` |
| 38 | Host string `json:"host"` |
| 39 | Entry json.RawMessage `json:"entry"` |
| 40 | } |
| 41 | |
| 42 | // outboxSegment refuses a path segment that is empty or could leave its |
| 43 | // directory. |
| 44 | func outboxSegment(s string) bool { |
| 45 | return s != "" && s != "." && s != ".." && !strings.ContainsAny(s, "/\\\x00") |
| 46 | } |
| 47 | |
| 48 | // queueLog keeps ev, for rec's app and host, in the outbox: written to a |
| 49 | // temporary file, synced, and linked into place create-exclusive. The first |
| 50 | // entry kept for an identity stays: a transition seen again after a restart |
| 51 | // (built again, later) never replaces its bytes or its time (fix round 3, |
| 52 | // ruling 4). |
| 53 | func (w *movesWorker) queueLog(rec provision.Record, ev moves.LogEvent) error { |
| 54 | if w.cfg.DataDir == "" { |
| 55 | return errors.New("no data directory for the move log's outbox") |
| 56 | } |
| 57 | if !outboxSegment(rec.Host) || !outboxSegment(ev.Op) || !outboxSegment(ev.Identity) { |
| 58 | return fmt.Errorf("an unusable move log entry name: %s/%s/%s", rec.Host, ev.Op, ev.Identity) |
| 59 | } |
| 60 | entry, err := json.Marshal(ev) |
| 61 | if err != nil { |
| 62 | return err |
| 63 | } |
| 64 | body, err := json.Marshal(moveLogPending{V: 1, Account: rec.Account, App: rec.App, Host: rec.Host, Entry: entry}) |
| 65 | if err != nil { |
| 66 | return err |
| 67 | } |
| 68 | _, err = outboxCreate(w.cfg.DataDir, filepath.Join(moveLogOutboxDir(w.cfg.DataDir), rec.Host, ev.Op), ev.Identity+".json", body) |
| 69 | return err |
| 70 | } |
| 71 | |
| 72 | // outboxCreate keeps body as dir/name, create-exclusive: written to a |
| 73 | // temporary file, synced, and linked into place, so a crash leaves the whole |
| 74 | // file or none. created is false, with no error, when dir/name was already |
| 75 | // there: the first one kept stays, and an existing file is found before |
| 76 | // anything is written, so storage that has turned full or read-only still |
| 77 | // answers for it (Plan 6 Tasks 4+5 fix round 3). The move log's outbox and |
| 78 | // the page outbox (movespaging.go) share it. |
| 79 | // |
| 80 | // base is the data directory the outbox lives under. Nothing is |
| 81 | // acknowledged until it is durable (final review I2): every directory the |
| 82 | // call creates is synced into its parent, and the entry's directory and |
| 83 | // each one above it up to base are synced after the link, any failure |
| 84 | // returned. An entry found already there may be an earlier try's whose sync |
| 85 | // failed after the link, so it counts only once it, its directory and those |
| 86 | // above it are synced now. |
| 87 | func outboxCreate(base, dir, name string, body []byte) (created bool, err error) { |
| 88 | if err := outboxBase(base); err != nil { |
| 89 | return false, err |
| 90 | } |
| 91 | path := filepath.Join(dir, name) |
| 92 | if _, err := os.Lstat(path); err == nil { |
| 93 | return false, outboxSettle(base, dir, path) |
| 94 | } |
| 95 | fresh, err := outboxMissingDirs(dir) |
| 96 | if err != nil { |
| 97 | return false, err |
| 98 | } |
| 99 | if err := os.MkdirAll(dir, 0o700); err != nil { |
| 100 | return false, err |
| 101 | } |
| 102 | for _, c := range fresh { |
| 103 | if err := outboxSyncPath(filepath.Dir(c)); err != nil { |
| 104 | return false, err |
| 105 | } |
| 106 | } |
| 107 | tmp, err := os.CreateTemp(dir, ".pending-*") |
| 108 | if err != nil { |
| 109 | return false, err |
| 110 | } |
| 111 | defer os.Remove(tmp.Name()) |
| 112 | if _, err := tmp.Write(body); err != nil { |
| 113 | tmp.Close() |
| 114 | return false, err |
| 115 | } |
| 116 | if err := tmp.Sync(); err != nil { |
| 117 | tmp.Close() |
| 118 | return false, err |
| 119 | } |
| 120 | if err := tmp.Close(); err != nil { |
| 121 | return false, err |
| 122 | } |
| 123 | if err := os.Link(tmp.Name(), path); err != nil { |
| 124 | if errors.Is(err, fs.ErrExist) { |
| 125 | return false, outboxSettle(base, dir, path) |
| 126 | } |
| 127 | return false, err |
| 128 | } |
| 129 | if err := outboxSyncChain(base, dir); err != nil { |
| 130 | return true, err |
| 131 | } |
| 132 | return true, nil |
| 133 | } |
| 134 | |
| 135 | // outboxBase refuses a data directory that is not absolute and clean: the |
| 136 | // ancestor walk (outboxSyncChain) ends only at such a root. The worker |
| 137 | // resolves its own once, at construction (movesDataDir). |
| 138 | func outboxBase(base string) error { |
| 139 | if !filepath.IsAbs(base) || filepath.Clean(base) != base { |
| 140 | return fmt.Errorf("the outbox's data directory %q is not an absolute, clean path", base) |
| 141 | } |
| 142 | return nil |
| 143 | } |
| 144 | |
| 145 | // outboxSync syncs one open file or directory of an outbox. It is a |
| 146 | // variable only so the host's tests can fail it; nothing in production |
| 147 | // configuration reaches it. |
| 148 | var outboxSync = func(f *os.File) error { return f.Sync() } |
| 149 | |
| 150 | // outboxSyncPath syncs the file or directory at p. |
| 151 | func outboxSyncPath(p string) error { |
| 152 | f, err := os.Open(p) |
| 153 | if err != nil { |
| 154 | return err |
| 155 | } |
| 156 | serr := outboxSync(f) |
| 157 | if cerr := f.Close(); serr == nil { |
| 158 | serr = cerr |
| 159 | } |
| 160 | return serr |
| 161 | } |
| 162 | |
| 163 | // outboxSyncChain syncs dir and each directory above it up to and |
| 164 | // including base, so the entries naming them are all on disk. A dir outside |
| 165 | // base is synced alone. |
| 166 | func outboxSyncChain(base, dir string) error { |
| 167 | base, dir = filepath.Clean(base), filepath.Clean(dir) |
| 168 | for p := dir; ; { |
| 169 | if err := outboxSyncPath(p); err != nil { |
| 170 | return err |
| 171 | } |
| 172 | q := filepath.Dir(p) |
| 173 | if p == base || q == p || !outboxWithin(base, q) { |
| 174 | return nil |
| 175 | } |
| 176 | p = q |
| 177 | } |
| 178 | } |
| 179 | |
| 180 | // outboxWithin reports whether p is base or under it. |
| 181 | func outboxWithin(base, p string) bool { |
| 182 | if base == "" || base == "." { |
| 183 | return false |
| 184 | } |
| 185 | rel, err := filepath.Rel(base, p) |
| 186 | return err == nil && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator)) |
| 187 | } |
| 188 | |
| 189 | // outboxSettle makes an entry found on disk durable before it counts: the |
| 190 | // file, its directory and those above it up to base. |
| 191 | func outboxSettle(base, dir, path string) error { |
| 192 | if err := outboxSyncPath(path); err != nil { |
| 193 | return err |
| 194 | } |
| 195 | return outboxSyncChain(base, dir) |
| 196 | } |
| 197 | |
| 198 | // outboxMissingDirs lists the directories at and above dir that do not |
| 199 | // exist yet, deepest first: the ones MkdirAll is about to create. |
| 200 | func outboxMissingDirs(dir string) ([]string, error) { |
| 201 | var out []string |
| 202 | for p := filepath.Clean(dir); ; p = filepath.Dir(p) { |
| 203 | _, err := os.Lstat(p) |
| 204 | if err == nil { |
| 205 | return out, nil |
| 206 | } |
| 207 | if !errors.Is(err, fs.ErrNotExist) { |
| 208 | return nil, err |
| 209 | } |
| 210 | out = append(out, p) |
| 211 | if filepath.Dir(p) == p { |
| 212 | return out, nil |
| 213 | } |
| 214 | } |
| 215 | } |
| 216 | |
| 217 | // outboxReplace replaces dir/name with body: a temporary file, synced and |
| 218 | // renamed over it, so a crash leaves the old file or the new one. Like |
| 219 | // outboxCreate, it succeeds only once the rename is durable: dir and each |
| 220 | // directory above it up to base synced (final review I2). |
| 221 | func outboxReplace(base, dir, name string, body []byte) error { |
| 222 | if err := outboxBase(base); err != nil { |
| 223 | return err |
| 224 | } |
| 225 | tmp, err := os.CreateTemp(dir, ".pending-*") |
| 226 | if err != nil { |
| 227 | return err |
| 228 | } |
| 229 | defer os.Remove(tmp.Name()) |
| 230 | if _, err := tmp.Write(body); err != nil { |
| 231 | tmp.Close() |
| 232 | return err |
| 233 | } |
| 234 | if err := tmp.Sync(); err != nil { |
| 235 | tmp.Close() |
| 236 | return err |
| 237 | } |
| 238 | if err := tmp.Close(); err != nil { |
| 239 | return err |
| 240 | } |
| 241 | if err := os.Rename(tmp.Name(), filepath.Join(dir, name)); err != nil { |
| 242 | return err |
| 243 | } |
| 244 | return outboxSyncChain(base, dir) |
| 245 | } |
| 246 | |
| 247 | // flushLogOutbox retries every kept entry through the LogEvent hook, into s. |
| 248 | // A written entry's file is deleted; one that fails again stays. A file that |
| 249 | // cannot be read as an entry is logged and deleted: it can never be written. |
| 250 | func (w *movesWorker) flushLogOutbox(ctx context.Context, s release.Store) { |
| 251 | f := w.cfg.LogEvent |
| 252 | if w.cfg.DataDir == "" || f == nil || s == nil { |
| 253 | return |
| 254 | } |
| 255 | root := moveLogOutboxDir(w.cfg.DataDir) |
| 256 | err := filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error { |
| 257 | if errors.Is(err, fs.ErrNotExist) && p == root { |
| 258 | return fs.SkipAll |
| 259 | } |
| 260 | if err != nil { |
| 261 | return err |
| 262 | } |
| 263 | if d.IsDir() || !strings.HasSuffix(d.Name(), ".json") || strings.HasPrefix(d.Name(), ".") { |
| 264 | return nil |
| 265 | } |
| 266 | if err := ctx.Err(); err != nil { |
| 267 | return err |
| 268 | } |
| 269 | w.flushLogFile(ctx, s, f, p) |
| 270 | return nil |
| 271 | }) |
| 272 | if err != nil { |
| 273 | log.Printf("moves: the move log's outbox: %v", err) |
| 274 | } |
| 275 | pruneEmptyDirs(root) |
| 276 | } |
| 277 | |
| 278 | // flushLogFile retries one kept entry. |
| 279 | func (w *movesWorker) flushLogFile(ctx context.Context, s release.Store, f func(context.Context, release.Store, provision.Record, moves.LogEvent) error, p string) { |
| 280 | body, err := os.ReadFile(p) |
| 281 | if err != nil { |
| 282 | log.Printf("moves: the move log's outbox: %v", err) |
| 283 | return |
| 284 | } |
| 285 | var kept moveLogPending |
| 286 | var ev moves.LogEvent |
| 287 | if err := json.Unmarshal(body, &kept); err != nil || kept.V != 1 || json.Unmarshal(kept.Entry, &ev) != nil || |
| 288 | ev.Identity+".json" != filepath.Base(p) || ev.Op != filepath.Base(filepath.Dir(p)) || kept.Host != filepath.Base(filepath.Dir(filepath.Dir(p))) { |
| 289 | log.Printf("moves: the move log's outbox: %s is not a kept entry; deleted", p) |
| 290 | _ = os.Remove(p) |
| 291 | return |
| 292 | } |
| 293 | if err := f(ctx, s, provision.Record{Account: kept.Account, App: kept.App, Host: kept.Host}, ev); err != nil { |
| 294 | log.Printf("moves: %s: the log of %s at %s is kept for the next pass: %v", kept.Host, ev.Op, ev.Identity, err) |
| 295 | return |
| 296 | } |
| 297 | if err := os.Remove(p); err != nil { |
| 298 | log.Printf("moves: the move log's outbox: %v", err) |
| 299 | } |
| 300 | } |
| 301 | |
| 302 | // pruneEmptyDirs removes the outbox's empty host and move directories. |
| 303 | func pruneEmptyDirs(root string) { |
| 304 | hosts, err := os.ReadDir(root) |
| 305 | if err != nil { |
| 306 | return |
| 307 | } |
| 308 | for _, h := range hosts { |
| 309 | if !h.IsDir() { |
| 310 | continue |
| 311 | } |
| 312 | hd := filepath.Join(root, h.Name()) |
| 313 | ops, _ := os.ReadDir(hd) |
| 314 | for _, o := range ops { |
| 315 | if o.IsDir() { |
| 316 | _ = os.Remove(filepath.Join(hd, o.Name())) // only if empty |
| 317 | } |
| 318 | } |
| 319 | _ = os.Remove(hd) |
| 320 | } |
| 321 | } |
| 322 | |
amadan