|
| 1 | +//go:build itest |
| 2 | + |
| 3 | +package itest |
| 4 | + |
| 5 | +import ( |
| 6 | + "bytes" |
| 7 | + "fmt" |
| 8 | + "strings" |
| 9 | + "testing" |
| 10 | + "time" |
| 11 | + |
| 12 | + ingottest "github.com/fil-forge/ingot/testing" |
| 13 | + "github.com/fil-forge/smelt/pkg/stack" |
| 14 | +) |
| 15 | + |
| 16 | +// TestForgeReadAfterCatalogRetention proves catalog retention behaves as a |
| 17 | +// cache, not an availability cliff: after a shipped catalog segment is retired |
| 18 | +// off local disk, its blocks (object manifests, MST nodes) must still resolve |
| 19 | +// through the fallthrough read tier — the shard_inclusions row (block → shard |
| 20 | +// CAR + byte range, recorded at ship) joined to the shard's blob_locations |
| 21 | +// row, retrieved as a ranged /content/retrieve against piri. |
| 22 | +// |
| 23 | +// The config seals the catalog every 1s, retains ONE shipped segment, and |
| 24 | +// disables the read cache. An early object is written, then later writes roll |
| 25 | +// the catalog past the retain window until the early object's segment is |
| 26 | +// retired; the early object must then still GET (its manifest is only in the |
| 27 | +// retired segment) and the bucket must still list without a delimiter (the |
| 28 | +// walk fetches every leaf's manifest). |
| 29 | +// |
| 30 | +// go test -tags itest ./itest -run TestForgeReadAfterCatalogRetention -v -timeout 900s |
| 31 | +func TestForgeReadAfterCatalogRetention(t *testing.T) { |
| 32 | + ctx := t.Context() |
| 33 | + |
| 34 | + s, ingotEndpoint := forgeStack(t, stack.WithServiceConfig("ingot", "testdata/config-retention.yaml")) |
| 35 | + accessKey, secretKey := hiltProvisionTenant(t, ctx, s, "retention") |
| 36 | + cfg := forgeConfig(ingotEndpoint, accessKey, secretKey) |
| 37 | + |
| 38 | + const bucket = "retention-bucket" |
| 39 | + if err := ingottest.CreateBucket(ctx, cfg, bucket); err != nil { |
| 40 | + t.Fatalf("create bucket: %v", err) |
| 41 | + } |
| 42 | + |
| 43 | + // The early object: its manifest lands in the first catalog segment(s), |
| 44 | + // which the later writes will push out of the retain window. |
| 45 | + early := patternBytes(64 << 10) |
| 46 | + if err := ingottest.PutBytes(ctx, cfg, bucket, "early-obj", early); err != nil { |
| 47 | + t.Fatalf("put early object: %v", err) |
| 48 | + } |
| 49 | + |
| 50 | + // Count the segment CARs currently on disk; the early manifest lives in |
| 51 | + // one of these. Retirement is proven when every one of them is gone. |
| 52 | + initialSegs := catalogSegments(t, s, bucket) |
| 53 | + if len(initialSegs) == 0 { |
| 54 | + // The first segment may not have sealed yet; wait for it so we have |
| 55 | + // a concrete set to watch retire. |
| 56 | + waitFor(t, time.Minute, "first catalog segment to seal", func() bool { |
| 57 | + initialSegs = catalogSegments(t, s, bucket) |
| 58 | + return len(initialSegs) > 0 |
| 59 | + }) |
| 60 | + } |
| 61 | + t.Logf("early object's manifest is in segment(s): %v", initialSegs) |
| 62 | + |
| 63 | + // Roll the catalog: spaced writes each seal (1s seal_age) and ship a new |
| 64 | + // segment; retain=1 retires everything older. Keep writing until every |
| 65 | + // initial segment file is gone from the container. |
| 66 | + waitFor(t, 5*time.Minute, "initial catalog segments to retire", func() bool { |
| 67 | + key := fmt.Sprintf("filler/obj-%d", time.Now().UnixNano()) |
| 68 | + if err := ingottest.PutBytes(ctx, cfg, bucket, key, patternBytes(4<<10)); err != nil { |
| 69 | + t.Fatalf("put filler object: %v", err) |
| 70 | + } |
| 71 | + // Slower than seal_age so each segment seals and ships with headroom |
| 72 | + // (no flush-queue overflow → no re-ships through the dedup path). |
| 73 | + time.Sleep(3 * time.Second) |
| 74 | + remaining := catalogSegments(t, s, bucket) |
| 75 | + for _, seg := range initialSegs { |
| 76 | + for _, r := range remaining { |
| 77 | + if seg == r { |
| 78 | + return false |
| 79 | + } |
| 80 | + } |
| 81 | + } |
| 82 | + return true |
| 83 | + }) |
| 84 | + t.Logf("initial segments retired; segments now on disk: %v", catalogSegments(t, s, bucket)) |
| 85 | + |
| 86 | + // GET the early object: its manifest exists only in a retired segment, so |
| 87 | + // this read MUST come through inclusion → shard → ranged piri retrieval. |
| 88 | + got, err := ingottest.GetBytes(ctx, cfg, bucket, "early-obj") |
| 89 | + if err != nil { |
| 90 | + t.Fatalf("get early object after its catalog segment retired: %v", err) |
| 91 | + } |
| 92 | + if !bytes.Equal(got, early) { |
| 93 | + t.Fatalf("early object mismatch after retention: got %d bytes, want %d", len(got), len(early)) |
| 94 | + } |
| 95 | + |
| 96 | + // Undelimited list walks every leaf and fetches every manifest — the |
| 97 | + // regression that motivated this test (root listing failed with |
| 98 | + // "blockstore: not found" once a manifest's segment was retired). |
| 99 | + keys, err := ingottest.ListKeys(ctx, cfg, bucket) |
| 100 | + if err != nil { |
| 101 | + t.Fatalf("list bucket after retention: %v", err) |
| 102 | + } |
| 103 | + found := false |
| 104 | + for _, k := range keys { |
| 105 | + if k == "early-obj" { |
| 106 | + found = true |
| 107 | + break |
| 108 | + } |
| 109 | + } |
| 110 | + if !found { |
| 111 | + t.Fatalf("early-obj missing from listing: %v", keys) |
| 112 | + } |
| 113 | + t.Logf("read-after-retention OK: %d bytes via ranged shard retrieval; %d keys listed", len(got), len(keys)) |
| 114 | +} |
| 115 | + |
| 116 | +// catalogSegments lists the catalog-plane CAR files currently on the ingot |
| 117 | +// container's disk for bucket (segments live under |
| 118 | +// /data/segments/<bucket>/catalog/). |
| 119 | +func catalogSegments(t *testing.T, s *stack.Stack, bucket string) []string { |
| 120 | + t.Helper() |
| 121 | + out, _, err := s.Exec(t.Context(), "ingot", "sh", "-c", |
| 122 | + "ls /data/segments/"+bucket+"/catalog/ 2>/dev/null | grep '\\.car$' || true") |
| 123 | + if err != nil { |
| 124 | + t.Fatalf("list catalog segments: %v", err) |
| 125 | + } |
| 126 | + fields := strings.Fields(strings.TrimSpace(out)) |
| 127 | + return fields |
| 128 | +} |
| 129 | + |
| 130 | +// waitFor polls cond (which may do work per attempt) until true or fatal after |
| 131 | +// timeout. |
| 132 | +func waitFor(t *testing.T, timeout time.Duration, what string, cond func() bool) { |
| 133 | + t.Helper() |
| 134 | + deadline := time.Now().Add(timeout) |
| 135 | + for time.Now().Before(deadline) { |
| 136 | + if cond() { |
| 137 | + return |
| 138 | + } |
| 139 | + } |
| 140 | + t.Fatalf("timed out after %s waiting for %s", timeout, what) |
| 141 | +} |
0 commit comments