| 1 | package layout |
| 2 | |
| 3 | import ( |
| 4 | "encoding/json" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "os" |
| 9 | "path/filepath" |
| 10 | "reflect" |
| 11 | "strings" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/carlosframework/platform/internal/provision" |
| 15 | "github.com/carlosframework/platform/internal/registry" |
| 16 | ) |
| 17 | |
| 18 | // Phase is how far a move got. files: sources are being renamed; row: the |
| 19 | // files are at their destinations and the row is being swapped; done: the |
| 20 | // row is swapped and only the journal remains. |
| 21 | type Phase string |
| 22 | |
| 23 | const ( |
| 24 | PhaseFiles Phase = "files" |
| 25 | PhaseRow Phase = "row" |
| 26 | PhaseDone Phase = "done" |
| 27 | ) |
| 28 | |
| 29 | // Journal is /data/layout/<host>.json (design §8.2 step 6). It records |
| 30 | // presence so recovery can tell an optional artifact that was absent from |
| 31 | // one that was lost (R2-7). It is NOT an instruction (§3.3): Validate |
| 32 | // re-derives both endpoints from the host and the mode and refuses a |
| 33 | // journal that names anything else. |
| 34 | type Journal struct { |
| 35 | Host string `json:"host"` |
| 36 | Mode Mode `json:"mode"` |
| 37 | From Endpoint `json:"from"` |
| 38 | To Endpoint `json:"to"` |
| 39 | Artifacts []JournalArtifact `json:"artifacts"` |
| 40 | // Extra is the app-made entries moved with the host (extras.go), by |
| 41 | // base name. Absent in journals written before #256, which moved |
| 42 | // none. Each is re-derived from the host and the name on recovery |
| 43 | // and must pass extraNameOK, so the journal cannot direct a move |
| 44 | // PlanFor and ScanExtras would not have chosen. |
| 45 | Extra []JournalArtifact `json:"extra,omitempty"` |
| 46 | Phase Phase `json:"phase"` |
| 47 | WrittenAt string `json:"written_at"` |
| 48 | } |
| 49 | |
| 50 | // extraNames is the recorded extra entries' names, in journal order. |
| 51 | func (j *Journal) extraNames() []string { |
| 52 | var names []string |
| 53 | for _, e := range j.Extra { |
| 54 | names = append(names, e.Name) |
| 55 | } |
| 56 | return names |
| 57 | } |
| 58 | |
| 59 | // JournalArtifact records one artifact's presence at the time the journal |
| 60 | // was written. The path itself is not stored: it is re-derived by |
| 61 | // PlanFor from the host and mode on every read, so a journal can never be |
| 62 | // used to redirect a rename anywhere PlanFor itself would not have chosen. |
| 63 | type JournalArtifact struct { |
| 64 | Name string `json:"name"` |
| 65 | Present bool `json:"present"` |
| 66 | } |
| 67 | |
| 68 | // ErrJournalExists is WriteJournal's answer when a journal is already on |
| 69 | // disk for the host. |
| 70 | var ErrJournalExists = errors.New("layout: a journal already exists for this host") |
| 71 | |
| 72 | // journalMax bounds how much of a journal file ReadJournal will read. A |
| 73 | // journal is a handful of paths and booleans; anything past this is either |
| 74 | // corrupt or something else entirely. |
| 75 | const journalMax = 64 << 10 |
| 76 | |
| 77 | func journalDir(dataDir string) string { return filepath.Join(dataDir, "layout") } |
| 78 | |
| 79 | // syncDir fsyncs a directory. A file's own Sync makes its bytes durable, |
| 80 | // but the directory entry that gives those bytes a name (a hard link, a |
| 81 | // rename) is separate metadata: on ext4 with delayed allocation, a crash |
| 82 | // between an otherwise-successful Link or Rename and this call can still |
| 83 | // lose the name while the bytes it would have pointed to survive under |
| 84 | // their old one. |
| 85 | func syncDir(dir string) error { |
| 86 | d, err := os.Open(dir) |
| 87 | if err != nil { |
| 88 | return err |
| 89 | } |
| 90 | defer d.Close() |
| 91 | return d.Sync() |
| 92 | } |
| 93 | |
| 94 | // WriteJournal creates the journal exactly once. An existing file, whoever |
| 95 | // wrote it, is a refusal: the pass never overwrites a record of a move it |
| 96 | // did not make. |
| 97 | // |
| 98 | // It writes the content to a temp file, syncs that file, then links it to |
| 99 | // the final name: os.Link fails with EEXIST exactly as O_EXCL would, |
| 100 | // without ever making a partly written journal visible at the final path. |
| 101 | // Without the temp+sync+link, an O_EXCL create-and-write can leave an |
| 102 | // empty file at the final name durable while its content is not (a power |
| 103 | // loss, or a kill between OpenFile and Write). Recovery then finds a |
| 104 | // journal with no content, refuses it, and the host stays fenced with |
| 105 | // nothing to read to explain why. |
| 106 | func WriteJournal(dataDir string, j *Journal) error { |
| 107 | dir := journalDir(dataDir) |
| 108 | if err := os.MkdirAll(dir, 0o700); err != nil { |
| 109 | return err |
| 110 | } |
| 111 | if j.WrittenAt == "" { |
| 112 | j.WrittenAt = time.Now().UTC().Format(time.RFC3339) |
| 113 | } |
| 114 | tmpName, err := writeTemp(dir, j) |
| 115 | if err != nil { |
| 116 | return err |
| 117 | } |
| 118 | linkErr := os.Link(tmpName, provision.LayoutJournalPath(dataDir, j.Host)) |
| 119 | // The temp file's only job was to become the final name; whether that |
| 120 | // succeeded or not, its own name must not linger (nothing reads a |
| 121 | // ".tmp-" name, but a stray one left after a failed Link would look |
| 122 | // like leftover corruption to anyone auditing the directory by hand). |
| 123 | os.Remove(tmpName) |
| 124 | if linkErr != nil { |
| 125 | if errors.Is(linkErr, os.ErrExist) { |
| 126 | return ErrJournalExists |
| 127 | } |
| 128 | return linkErr |
| 129 | } |
| 130 | return syncDir(dir) |
| 131 | } |
| 132 | |
| 133 | // ReadJournal reads the host's journal without following a symlink and |
| 134 | // without blocking on a special file: /data/layout is agent-owned, but a |
| 135 | // flat tenant shares the agent's uid during the transition and can plant |
| 136 | // things there. |
| 137 | func ReadJournal(dataDir, host string) (*Journal, bool, error) { |
| 138 | path := provision.LayoutJournalPath(dataDir, host) |
| 139 | f, err := os.OpenFile(path, os.O_RDONLY|syscallNofollow|syscallNonblock, 0) |
| 140 | if errors.Is(err, os.ErrNotExist) { |
| 141 | return nil, false, nil |
| 142 | } |
| 143 | if err != nil { |
| 144 | return nil, false, err |
| 145 | } |
| 146 | defer f.Close() |
| 147 | fi, err := f.Stat() |
| 148 | if err != nil { |
| 149 | return nil, false, err |
| 150 | } |
| 151 | if !fi.Mode().IsRegular() { |
| 152 | return nil, false, fmt.Errorf("layout: %s is not a regular file", path) |
| 153 | } |
| 154 | raw, err := io.ReadAll(io.LimitReader(f, journalMax+1)) |
| 155 | if err != nil { |
| 156 | return nil, false, err |
| 157 | } |
| 158 | if len(raw) > journalMax { |
| 159 | return nil, false, fmt.Errorf("layout: %s is larger than %d bytes", path, journalMax) |
| 160 | } |
| 161 | var j Journal |
| 162 | if err := json.Unmarshal(raw, &j); err != nil { |
| 163 | return nil, false, fmt.Errorf("layout: %s: %w", path, err) |
| 164 | } |
| 165 | if j.Host != host { |
| 166 | return nil, false, fmt.Errorf("layout: %s: content names host %q, path names %q", path, j.Host, host) |
| 167 | } |
| 168 | return &j, true, nil |
| 169 | } |
| 170 | |
| 171 | // UpdatePhase rewrites the journal with a new phase through a temp file and |
| 172 | // rename in the same directory, so a crash leaves either the old phase or |
| 173 | // the new one, never a torn file. It marshals a copy carrying the new |
| 174 | // phase, not j itself, so a caller that inspects j after an error sees the |
| 175 | // phase the journal on disk actually has, not one that was never made |
| 176 | // durable. |
| 177 | func UpdatePhase(dataDir string, j *Journal, p Phase) error { |
| 178 | next := *j |
| 179 | next.Phase = p |
| 180 | dir := journalDir(dataDir) |
| 181 | name, err := writeTemp(dir, &next) |
| 182 | if err != nil { |
| 183 | return err |
| 184 | } |
| 185 | if err := os.Rename(name, provision.LayoutJournalPath(dataDir, j.Host)); err != nil { |
| 186 | os.Remove(name) |
| 187 | return err |
| 188 | } |
| 189 | // Only now, with the rename itself having succeeded, does the phase |
| 190 | // this call promised become true of both the in-memory journal and |
| 191 | // the one on disk. |
| 192 | j.Phase = p |
| 193 | return syncDir(dir) |
| 194 | } |
| 195 | |
| 196 | // writeTemp writes j to a fresh temp file in dir and syncs it, returning |
| 197 | // its name. The sync comes before the caller links or renames the file to |
| 198 | // the journal's canonical name: that makes these bytes visible there, so |
| 199 | // they must already be durable, or a crash just after could leave the name |
| 200 | // pointing at content that never landed. |
| 201 | func writeTemp(dir string, j *Journal) (string, error) { |
| 202 | raw, err := json.MarshalIndent(j, "", " ") |
| 203 | if err != nil { |
| 204 | return "", err |
| 205 | } |
| 206 | tmp, err := os.CreateTemp(dir, "."+j.Host+".tmp-*") |
| 207 | if err != nil { |
| 208 | return "", err |
| 209 | } |
| 210 | name := tmp.Name() |
| 211 | if _, err := tmp.Write(append(raw, '\n')); err != nil { |
| 212 | tmp.Close() |
| 213 | os.Remove(name) |
| 214 | return "", err |
| 215 | } |
| 216 | if err := tmp.Sync(); err != nil { |
| 217 | tmp.Close() |
| 218 | os.Remove(name) |
| 219 | return "", err |
| 220 | } |
| 221 | if err := tmp.Close(); err != nil { |
| 222 | os.Remove(name) |
| 223 | return "", err |
| 224 | } |
| 225 | return name, nil |
| 226 | } |
| 227 | |
| 228 | // ReplaceJournal hands a host over from the journal the caller holds to a |
| 229 | // new one in a single rename, so the host is never without a journal: a |
| 230 | // crash, or a failure anywhere here, leaves either the whole old journal |
| 231 | // or the whole new one at the canonical name. Removing the old one and |
| 232 | // then creating the new one would leave a gap in which nothing fences the |
| 233 | // host, and a same-uid tenant that can make <data>/layout unwritable at |
| 234 | // the right moment could hold that gap open. |
| 235 | // |
| 236 | // The file replaced must be exactly held, as ReadJournal reads it back; |
| 237 | // anything else there is refused and left alone. No exclusive create is |
| 238 | // needed on the final name: the caller holds both the host's fence and |
| 239 | // the section config sync is locked out of, and it has just checked what |
| 240 | // it is replacing. |
| 241 | func ReplaceJournal(dataDir string, held, next *Journal) error { |
| 242 | if next.Host != held.Host { |
| 243 | return fmt.Errorf("layout: replacing %s's journal with one for %s", held.Host, next.Host) |
| 244 | } |
| 245 | cur, found, err := ReadJournal(dataDir, held.Host) |
| 246 | if err != nil { |
| 247 | return err |
| 248 | } |
| 249 | if !found { |
| 250 | return fmt.Errorf("layout: %s: no journal to replace", held.Host) |
| 251 | } |
| 252 | if !reflect.DeepEqual(cur, held) { |
| 253 | return fmt.Errorf("layout: %s: the journal on disk is not the one this pass holds; not replacing it", held.Host) |
| 254 | } |
| 255 | if next.WrittenAt == "" { |
| 256 | next.WrittenAt = time.Now().UTC().Format(time.RFC3339) |
| 257 | } |
| 258 | dir := journalDir(dataDir) |
| 259 | name, err := writeTemp(dir, next) |
| 260 | if err != nil { |
| 261 | return err |
| 262 | } |
| 263 | if err := os.Rename(name, provision.LayoutJournalPath(dataDir, next.Host)); err != nil { |
| 264 | os.Remove(name) |
| 265 | return err |
| 266 | } |
| 267 | return syncDir(dir) |
| 268 | } |
| 269 | |
| 270 | // RemoveJournal deletes the host's journal. Removing an absent journal is |
| 271 | // not an error: callers reach this after a move is already complete, and a |
| 272 | // second call (a retried recovery, say) must be idempotent. |
| 273 | func RemoveJournal(dataDir, host string) error { |
| 274 | err := os.Remove(provision.LayoutJournalPath(dataDir, host)) |
| 275 | if errors.Is(err, os.ErrNotExist) { |
| 276 | return nil |
| 277 | } |
| 278 | return err |
| 279 | } |
| 280 | |
| 281 | // JournalHosts names every host with a journal in <data>/layout, in the |
| 282 | // directory's order. Dot names are WriteJournal's and UpdatePhase's temp |
| 283 | // files, and are not journals. A directory that is not there holds none. |
| 284 | func JournalHosts(dataDir string) ([]string, error) { |
| 285 | entries, err := os.ReadDir(journalDir(dataDir)) |
| 286 | if errors.Is(err, os.ErrNotExist) { |
| 287 | return nil, nil |
| 288 | } |
| 289 | if err != nil { |
| 290 | return nil, err |
| 291 | } |
| 292 | var hosts []string |
| 293 | for _, e := range entries { |
| 294 | if n := e.Name(); !strings.HasPrefix(n, ".") && strings.HasSuffix(n, ".json") { |
| 295 | hosts = append(hosts, strings.TrimSuffix(n, ".json")) |
| 296 | } |
| 297 | } |
| 298 | return hosts, nil |
| 299 | } |
| 300 | |
| 301 | // JournalPresent reports whether a journal exists for host, without |
| 302 | // reading it. Later tasks use this everywhere a caller needs to know a |
| 303 | // host is "fenced" by an in-flight move but has no reason to read the |
| 304 | // journal's contents (design §8.1's LayoutFenced, SyncExec's skip, the |
| 305 | // litestream fenced hook). An Lstat error other than "not exist" (a |
| 306 | // permission problem, say) is treated as present: silently proceeding |
| 307 | // past a journal this process cannot even stat is worse than refusing. |
| 308 | func JournalPresent(dataDir, host string) bool { |
| 309 | return provision.LayoutJournalExists(dataDir, host) |
| 310 | } |
| 311 | |
| 312 | // Validate is the recovery check of design §8.2: the journal's endpoints |
| 313 | // must be exactly what PlanFor derives for its host and mode, and the row |
| 314 | // must sit at one of them. It returns whether the row is at `from` (the |
| 315 | // CAS has not run) or at `to` (it has). |
| 316 | func Validate(j *Journal, dataDir, socketDir string, row registry.Route) (atFrom bool, err error) { |
| 317 | if j.Host != row.Host { |
| 318 | return false, fmt.Errorf("layout: journal is for %s, row is %s", j.Host, row.Host) |
| 319 | } |
| 320 | switch j.Phase { |
| 321 | case PhaseFiles, PhaseRow, PhaseDone: |
| 322 | default: |
| 323 | return false, fmt.Errorf("layout: %s: journal phase %q is not files, row or done; not acting", j.Host, j.Phase) |
| 324 | } |
| 325 | p, err := PlanFor(dataDir, socketDir, j.Host, j.Mode) |
| 326 | if err != nil { |
| 327 | return false, err |
| 328 | } |
| 329 | if j.From != p.From || j.To != p.To { |
| 330 | return false, fmt.Errorf("layout: %s: journal endpoints %+v -> %+v are not the derived %+v -> %+v; not acting", j.Host, j.From, j.To, p.From, p.To) |
| 331 | } |
| 332 | if len(j.Artifacts) != len(artifactNames) { |
| 333 | return false, fmt.Errorf("layout: %s: journal lists %d artifacts, expected %d", j.Host, len(j.Artifacts), len(artifactNames)) |
| 334 | } |
| 335 | for i, a := range j.Artifacts { |
| 336 | if a.Name != artifactNames[i] { |
| 337 | return false, fmt.Errorf("layout: %s: journal artifact %d is %q, expected %q", j.Host, i, a.Name, artifactNames[i]) |
| 338 | } |
| 339 | } |
| 340 | // The db is the one artifact recovery cannot proceed without: every |
| 341 | // other artifact missing is a smaller, optional file the pass would |
| 342 | // simply skip, but a journal claiming its db was never even there |
| 343 | // describes a move that cannot have happened. |
| 344 | if !j.Artifacts[0].Present { |
| 345 | return false, fmt.Errorf("layout: %s: journal's db artifact is not marked present; not acting", j.Host) |
| 346 | } |
| 347 | // The extra names pass the same rule a scan applies, and an extra is |
| 348 | // never absent: the plan carries one because it was seen. |
| 349 | if err := p.AddExtras(j.extraNames()); err != nil { |
| 350 | return false, fmt.Errorf("%w; not acting", err) |
| 351 | } |
| 352 | for _, e := range j.Extra { |
| 353 | if !e.Present { |
| 354 | return false, fmt.Errorf("layout: %s: journal records extra %q absent; not acting", j.Host, e.Name) |
| 355 | } |
| 356 | } |
| 357 | switch (Endpoint{DB: row.DBPath, Socket: row.Socket}) { |
| 358 | case j.From: |
| 359 | return true, nil |
| 360 | case j.To: |
| 361 | return false, nil |
| 362 | } |
| 363 | return false, fmt.Errorf("layout: %s: row holds db %q socket %q, at neither journal endpoint; not acting", j.Host, row.DBPath, row.Socket) |
| 364 | } |
| 365 | |
amadan