| 1 | //go:build linux |
| 2 | |
| 3 | package host |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "errors" |
| 8 | "os" |
| 9 | "path/filepath" |
| 10 | "reflect" |
| 11 | "strings" |
| 12 | "testing" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/hostbox" |
| 16 | "github.com/carlosframework/platform/internal/provision" |
| 17 | "github.com/carlosframework/platform/internal/rootstate" |
| 18 | "golang.org/x/sys/unix" |
| 19 | ) |
| 20 | |
| 21 | func servingCollectionFixture(t *testing.T, legacy bool) (*servingParkDiskFixture, servingParkAuthorization, servingParkAuthorization) { |
| 22 | t.Helper() |
| 23 | f := newServingParkDiskFixture(t) |
| 24 | f.put("", "old served bytes") |
| 25 | if !legacy { |
| 26 | f.put(".gen", "G10") |
| 27 | } |
| 28 | old, receipts := f.evidence(10, legacy, "started") |
| 29 | if _, err := f.park(old, receipts); err != nil { |
| 30 | t.Fatal(err) |
| 31 | } |
| 32 | f.complete(old) |
| 33 | f.put("", "new served bytes") |
| 34 | f.put(".gen", "G11") |
| 35 | next, receipts := f.evidence(11, false, "started") |
| 36 | if _, err := f.park(next, receipts); err != nil { |
| 37 | t.Fatal(err) |
| 38 | } |
| 39 | f.complete(next) |
| 40 | return f, old, next |
| 41 | } |
| 42 | |
| 43 | func TestServingParkCollectionWindowAndReplay(t *testing.T) { |
| 44 | f, old, next := servingCollectionFixture(t, false) |
| 45 | if err := f.ledger.recoverCompleted(t.Context(), old.ID); err != nil { |
| 46 | t.Fatal(err) |
| 47 | } |
| 48 | r, err := f.ledger.read(old.ID) |
| 49 | if err != nil || !r.CollectAfter.Equal(f.now.Add(hostbox.JournalReceiptRetention)) { |
| 50 | t.Fatalf("missing retention deadline: %+v %v", r, err) |
| 51 | } |
| 52 | before, _ := f.disk.Read("retirement-" + old.ID + ".json") |
| 53 | for _, now := range []time.Time{f.now.Add(-time.Hour), r.CollectAfter.Add(-time.Nanosecond)} { |
| 54 | f.now = now |
| 55 | if err = f.ledger.recoverCompleted(t.Context(), old.ID); err != nil { |
| 56 | t.Fatal(err) |
| 57 | } |
| 58 | after, err := f.disk.Read("retirement-" + old.ID + ".json") |
| 59 | if err != nil || string(before) != string(after) { |
| 60 | t.Fatal("deadline refreshed or record collected early", err) |
| 61 | } |
| 62 | } |
| 63 | // A restarted ledger uses only the persisted deadline. |
| 64 | restarted := *f.ledger |
| 65 | f.now = r.CollectAfter |
| 66 | if err = f.disk.Write("retirement-"+old.ID+".lock", nil); err != nil { |
| 67 | t.Fatal(err) |
| 68 | } |
| 69 | if err = restarted.recoverCompleted(t.Context(), old.ID); err != nil { |
| 70 | t.Fatal(err) |
| 71 | } |
| 72 | for _, suffix := range []string{".json", ".lock"} { |
| 73 | if _, err = f.disk.Read("retirement-" + old.ID + suffix); !errors.Is(err, os.ErrNotExist) { |
| 74 | t.Fatal("collected file survived", suffix, err) |
| 75 | } |
| 76 | } |
| 77 | f.retained(next, "new served bytes", r.ReapAfter) |
| 78 | // Even a fresh transport nonce must not rebuild an old collected intent |
| 79 | // from the current source. Root additionally checks the live claim. |
| 80 | f.put("", "current database") |
| 81 | f.put(".gen", "G10") |
| 82 | if _, err = f.park(old, r.Receipts); !errors.Is(err, hostbox.ErrChanged) { |
| 83 | t.Fatal("collected replay regained authority", err) |
| 84 | } |
| 85 | if _, err = f.ledger.read(old.ID); !errors.Is(err, os.ErrNotExist) { |
| 86 | t.Fatal("replay recreated retirement", err) |
| 87 | } |
| 88 | b, err := os.ReadFile(provision.NestedDBPath(f.data, f.base.Host)) |
| 89 | if err != nil || string(b) != "current database" { |
| 90 | t.Fatal("refused replay changed live source", err) |
| 91 | } |
| 92 | } |
| 93 | |
| 94 | func TestServingParkCollectionPreservesSlotReferences(t *testing.T) { |
| 95 | for _, reference := range []string{"ID", "Pending", "Last"} { |
| 96 | t.Run(reference, func(t *testing.T) { |
| 97 | f, old, _ := servingCollectionFixture(t, false) |
| 98 | if err := f.ledger.recoverCompleted(t.Context(), old.ID); err != nil { |
| 99 | t.Fatal(err) |
| 100 | } |
| 101 | r, _ := f.ledger.read(old.ID) |
| 102 | f.now = r.CollectAfter.Add(time.Hour) |
| 103 | slot, err := f.ledger.slot(old) |
| 104 | if err != nil { |
| 105 | t.Fatal(err) |
| 106 | } |
| 107 | switch reference { |
| 108 | case "ID": |
| 109 | slot.ID = old.ID |
| 110 | case "Pending": |
| 111 | slot.Pending = old.ID |
| 112 | case "Last": |
| 113 | slot.Last = old.ID |
| 114 | } |
| 115 | if err = f.ledger.withNamespace(t.Context(), func() error { return f.ledger.collectClosedLocked(t.Context(), &r, slot) }); err != nil { |
| 116 | t.Fatal(err) |
| 117 | } |
| 118 | if _, err = f.ledger.read(old.ID); err != nil { |
| 119 | t.Fatal("collected a referenced retirement", err) |
| 120 | } |
| 121 | }) |
| 122 | } |
| 123 | } |
| 124 | |
| 125 | func TestServingParkCollectionKeepsLegacyConsumption(t *testing.T) { |
| 126 | f, old, next := servingCollectionFixture(t, true) |
| 127 | consumed, err := f.ledger.legacyConsumption(old.Successor) |
| 128 | if err != nil || consumed == nil { |
| 129 | t.Fatal(err) |
| 130 | } |
| 131 | if err = f.ledger.recoverCompleted(t.Context(), old.ID); err != nil { |
| 132 | t.Fatal(err) |
| 133 | } |
| 134 | r, _ := f.ledger.read(old.ID) |
| 135 | f.now = r.CollectAfter |
| 136 | if err = f.ledger.recoverCompleted(t.Context(), old.ID); err != nil { |
| 137 | t.Fatal(err) |
| 138 | } |
| 139 | after, err := f.ledger.legacyConsumption(old.Successor) |
| 140 | if err != nil || after == nil || !after.Collected { |
| 141 | t.Fatal("collection lost the permanent refusal", err) |
| 142 | } |
| 143 | after.Collected = false |
| 144 | if !reflect.DeepEqual(after, consumed) { |
| 145 | t.Fatal("collection changed legacy evidence") |
| 146 | } |
| 147 | f.retained(next, "new served bytes", r.ReapAfter) |
| 148 | } |
| 149 | |
| 150 | func TestServingParkCollectionLegacyRootRefusesRecapture(t *testing.T) { |
| 151 | f, _ := newServingLegacyRootFixture(t, false) |
| 152 | a, err := f.p.parkLegacyServingDatabase(t.Context(), f.conn, f.current, servingLegacyPredecessor{}) |
| 153 | if err != nil { |
| 154 | t.Fatal(err) |
| 155 | } |
| 156 | c, err := f.p.ledger.legacyConsumption(f.current) |
| 157 | if err != nil || c == nil { |
| 158 | t.Fatal(err) |
| 159 | } |
| 160 | c.Collected = true |
| 161 | if err = f.p.ledger.write(servingLegacyKey("legacy", f.current), c); err != nil { |
| 162 | t.Fatal(err) |
| 163 | } |
| 164 | if err = f.disk.Remove("retirement-" + a.ID + ".json"); err != nil { |
| 165 | t.Fatal(err) |
| 166 | } |
| 167 | if _, err = f.p.parkLegacyServingDatabase(t.Context(), f.conn, f.current, servingLegacyPredecessor{}); !errors.Is(err, hostbox.ErrChanged) { |
| 168 | t.Fatal("root accepted collected legacy source", err) |
| 169 | } |
| 170 | if _, err = f.p.ledger.read(a.ID); !errors.Is(err, os.ErrNotExist) { |
| 171 | t.Fatal("root recreated collected record", err) |
| 172 | } |
| 173 | } |
| 174 | |
| 175 | func TestServingParkCollectionLegacyConsumptionBeforeIntentStillResumes(t *testing.T) { |
| 176 | f, _ := newServingLegacyRootFixture(t, false) |
| 177 | store := &servingCollectionFaultStore{Store: f.disk} |
| 178 | f.p.ledger.state = store |
| 179 | store.write = func(name string, body []byte) error { |
| 180 | if strings.HasPrefix(name, "retirement-") { |
| 181 | return unix.EIO |
| 182 | } |
| 183 | return f.disk.Write(name, body) |
| 184 | } |
| 185 | if _, err := f.p.parkLegacyServingDatabase(t.Context(), f.conn, f.current, servingLegacyPredecessor{}); !errors.Is(err, unix.EIO) { |
| 186 | t.Fatal("intent fault was not reached", err) |
| 187 | } |
| 188 | c, err := f.p.ledger.legacyConsumption(f.current) |
| 189 | if err != nil || c == nil || c.Collected { |
| 190 | t.Fatal("pending consumption unavailable", err) |
| 191 | } |
| 192 | if _, err := f.p.ledger.read(c.Authorization.ID); !errors.Is(err, os.ErrNotExist) { |
| 193 | t.Fatal("intent was saved", err) |
| 194 | } |
| 195 | f.p.ledger.state = f.disk |
| 196 | if _, err := f.p.parkLegacyServingDatabase(t.Context(), f.conn, f.current, servingLegacyPredecessor{}); err != nil { |
| 197 | t.Fatal("interrupted first capture did not resume", err) |
| 198 | } |
| 199 | } |
| 200 | |
| 201 | func TestServingParkCollectionRejectsInvalidDeadline(t *testing.T) { |
| 202 | f, old, _ := servingCollectionFixture(t, false) |
| 203 | r, err := f.ledger.read(old.ID) |
| 204 | if err != nil { |
| 205 | t.Fatal(err) |
| 206 | } |
| 207 | r.CollectAfter = f.now.Add(hostbox.JournalReceiptRetention) |
| 208 | for _, change := range []string{"incomplete", "open", "event-present", "namespace-unknown", "early"} { |
| 209 | t.Run(change, func(t *testing.T) { |
| 210 | bad := r |
| 211 | switch change { |
| 212 | case "incomplete": |
| 213 | bad.Complete = false |
| 214 | case "open": |
| 215 | bad.Closed = false |
| 216 | case "event-present": |
| 217 | bad.EventRemoved = false |
| 218 | case "namespace-unknown": |
| 219 | bad.Namespace = servingArtifactDirectory{} |
| 220 | case "early": |
| 221 | bad.CollectAfter = bad.ParkedAt.Add(hostbox.JournalReceiptRetention - time.Nanosecond) |
| 222 | } |
| 223 | if err := f.ledger.save(bad); !errors.Is(err, hostbox.ErrChanged) { |
| 224 | t.Fatal("invalid collection authority accepted", err) |
| 225 | } |
| 226 | }) |
| 227 | } |
| 228 | } |
| 229 | |
| 230 | type servingCollectionFaultStore struct { |
| 231 | rootstate.Store |
| 232 | remove func(string) error |
| 233 | write func(string, []byte) error |
| 234 | } |
| 235 | |
| 236 | func (s *servingCollectionFaultStore) Remove(name string) error { |
| 237 | if s.remove != nil { |
| 238 | return s.remove(name) |
| 239 | } |
| 240 | return s.Store.Remove(name) |
| 241 | } |
| 242 | |
| 243 | func (s *servingCollectionFaultStore) Write(name string, body []byte) error { |
| 244 | if s.write != nil { |
| 245 | return s.write(name, body) |
| 246 | } |
| 247 | return s.Store.Write(name, body) |
| 248 | } |
| 249 | |
| 250 | func TestServingParkCollectionInterruptedWrites(t *testing.T) { |
| 251 | for _, boundary := range []string{"deadline-before", "deadline-after", "consumption-before", "consumption-after", "lock-before", "lock-after", "record-before", "record-after"} { |
| 252 | t.Run(boundary, func(t *testing.T) { |
| 253 | f, old, _ := servingCollectionFixture(t, true) |
| 254 | deadline := strings.HasPrefix(boundary, "deadline") |
| 255 | if !deadline { |
| 256 | if err := f.ledger.recoverCompleted(t.Context(), old.ID); err != nil { |
| 257 | t.Fatal(err) |
| 258 | } |
| 259 | r, _ := f.ledger.read(old.ID) |
| 260 | f.now = r.CollectAfter |
| 261 | } |
| 262 | store := &servingCollectionFaultStore{Store: f.disk} |
| 263 | f.ledger.state = store |
| 264 | fired := false |
| 265 | fault := func(name string, apply func() error) error { |
| 266 | matches := deadline && strings.HasPrefix(name, "retirement-") && strings.HasSuffix(name, ".json") || strings.HasPrefix(boundary, "consumption") && strings.HasPrefix(name, "legacy-") || strings.HasPrefix(boundary, "lock") && strings.HasSuffix(name, ".lock") || strings.HasPrefix(boundary, "record") && strings.HasPrefix(name, "retirement-") && strings.HasSuffix(name, ".json") |
| 267 | if !matches || fired { |
| 268 | return apply() |
| 269 | } |
| 270 | fired = true |
| 271 | if strings.HasSuffix(boundary, "after") { |
| 272 | if err := apply(); err != nil && !errors.Is(err, os.ErrNotExist) { |
| 273 | return err |
| 274 | } |
| 275 | } |
| 276 | return unix.EIO |
| 277 | } |
| 278 | store.write = func(name string, body []byte) error { |
| 279 | return fault(name, func() error { return f.disk.Write(name, body) }) |
| 280 | } |
| 281 | store.remove = func(name string) error { return fault(name, func() error { return f.disk.Remove(name) }) } |
| 282 | if err := f.ledger.recoverCompleted(t.Context(), old.ID); !errors.Is(err, unix.EIO) || !fired { |
| 283 | t.Fatalf("fault not reached: fired=%t err=%v", fired, err) |
| 284 | } |
| 285 | f.ledger.state = f.disk |
| 286 | if boundary != "record-after" { |
| 287 | if err := f.ledger.recoverCompleted(t.Context(), old.ID); err != nil { |
| 288 | t.Fatal("restart did not resume", err) |
| 289 | } |
| 290 | } |
| 291 | if deadline { |
| 292 | r, err := f.ledger.read(old.ID) |
| 293 | if err != nil || !r.CollectAfter.Equal(f.now.Add(hostbox.JournalReceiptRetention)) { |
| 294 | t.Fatal("deadline lost after restart", err) |
| 295 | } |
| 296 | } else { |
| 297 | if _, err := f.ledger.read(old.ID); !errors.Is(err, os.ErrNotExist) { |
| 298 | t.Fatal("collection incomplete", err) |
| 299 | } |
| 300 | consumed, err := f.ledger.legacyConsumption(old.Successor) |
| 301 | if err != nil || consumed == nil || !consumed.Collected { |
| 302 | t.Fatal("legacy evidence lost", err) |
| 303 | } |
| 304 | } |
| 305 | }) |
| 306 | } |
| 307 | } |
| 308 | |
| 309 | func TestServingParkCollectionRefusesChangedFilesystem(t *testing.T) { |
| 310 | for _, change := range []string{"event", "namespace", "cancel"} { |
| 311 | t.Run(change, func(t *testing.T) { |
| 312 | f, old, _ := servingCollectionFixture(t, false) |
| 313 | if err := f.ledger.recoverCompleted(t.Context(), old.ID); err != nil { |
| 314 | t.Fatal(err) |
| 315 | } |
| 316 | r, _ := f.ledger.read(old.ID) |
| 317 | f.now = r.CollectAfter |
| 318 | ctx := t.Context() |
| 319 | var err error |
| 320 | switch change { |
| 321 | case "event": |
| 322 | err = os.Mkdir(filepath.Dir(f.path(old, "")), 0700) |
| 323 | case "namespace": |
| 324 | path := filepath.Join(f.data, "instances", ".serving-parked") |
| 325 | err = os.Rename(path, path+"-moved") |
| 326 | if err == nil { |
| 327 | err = os.Mkdir(path, 0700) |
| 328 | } |
| 329 | case "cancel": |
| 330 | var cancel context.CancelFunc |
| 331 | ctx, cancel = context.WithCancel(ctx) |
| 332 | cancel() |
| 333 | } |
| 334 | if err != nil { |
| 335 | t.Fatal(err) |
| 336 | } |
| 337 | before, _ := f.disk.Read("retirement-" + old.ID + ".json") |
| 338 | if err = f.ledger.recoverCompleted(ctx, old.ID); err == nil { |
| 339 | t.Fatal("collected through changed evidence") |
| 340 | } |
| 341 | after, err := f.disk.Read("retirement-" + old.ID + ".json") |
| 342 | if err != nil || string(before) != string(after) { |
| 343 | t.Fatal("refusal changed retirement", err) |
| 344 | } |
| 345 | }) |
| 346 | } |
| 347 | } |
| 348 | |
| 349 | func TestServingParkCollectionRebootAdjuncts(t *testing.T) { |
| 350 | for _, failure := range []string{"none", "preparation-before", "preparation-after", "binding-before", "binding-after"} { |
| 351 | t.Run(failure, func(t *testing.T) { |
| 352 | f, old, next := servingCollectionFixture(t, false) |
| 353 | if err := f.ledger.recoverCompleted(t.Context(), old.ID); err != nil { |
| 354 | t.Fatal(err) |
| 355 | } |
| 356 | r, err := f.ledger.read(old.ID) |
| 357 | if err != nil { |
| 358 | t.Fatal(err) |
| 359 | } |
| 360 | f.now = r.CollectAfter |
| 361 | // Collection treats preparation contents as inert private metadata. |
| 362 | prep := "reboot-" + old.ID + "-" + servingOtherBoot + ".json" |
| 363 | other := "reboot-" + next.ID + "-" + servingOtherBoot + ".json" |
| 364 | binding := "current-retirement-" + old.ID + ".json" |
| 365 | for _, name := range []string{prep, other} { |
| 366 | if err := f.disk.Write(name, []byte("{}")); err != nil { |
| 367 | t.Fatal(err) |
| 368 | } |
| 369 | } |
| 370 | fired := false |
| 371 | fault := &servingCollectionFaultStore{Store: f.disk} |
| 372 | fault.remove = func(name string) error { |
| 373 | match := strings.HasPrefix(failure, "preparation") && name == prep || strings.HasPrefix(failure, "binding") && name == binding |
| 374 | if match && !fired { |
| 375 | fired = true |
| 376 | if strings.HasSuffix(failure, "after") { |
| 377 | _ = f.disk.Remove(name) |
| 378 | } |
| 379 | return unix.EIO |
| 380 | } |
| 381 | return f.disk.Remove(name) |
| 382 | } |
| 383 | f.ledger.state = fault |
| 384 | err = f.ledger.recoverCompleted(t.Context(), old.ID) |
| 385 | if failure == "none" { |
| 386 | if err != nil { |
| 387 | t.Fatal(err) |
| 388 | } |
| 389 | } else if !fired || !errors.Is(err, unix.EIO) { |
| 390 | t.Fatalf("fault not reached: %v", err) |
| 391 | } |
| 392 | f.ledger.state = f.disk |
| 393 | if _, err := f.disk.Read("retirement-" + old.ID + ".json"); err == nil { |
| 394 | if err := f.ledger.recoverCompleted(t.Context(), old.ID); err != nil { |
| 395 | t.Fatal(err) |
| 396 | } |
| 397 | } |
| 398 | // A lost record-removal reply leaves only a binding; maintenance resumes it. |
| 399 | if err := f.disk.Write(binding, []byte("{}")); err != nil { |
| 400 | t.Fatal(err) |
| 401 | } |
| 402 | if err := f.ledger.maintainParks(t.Context(), func() error { return nil }); err != nil { |
| 403 | t.Fatal(err) |
| 404 | } |
| 405 | for _, name := range []string{prep, binding, "retirement-" + old.ID + ".json"} { |
| 406 | if _, err := f.disk.Read(name); !errors.Is(err, os.ErrNotExist) { |
| 407 | t.Fatalf("adjunct survived: %s %v", name, err) |
| 408 | } |
| 409 | } |
| 410 | if _, err := f.disk.Read(other); err != nil { |
| 411 | t.Fatal("collected another retirement's preparation", err) |
| 412 | } |
| 413 | }) |
| 414 | } |
| 415 | } |
| 416 | |
amadan