| 1 | package lease |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "errors" |
| 7 | "io" |
| 8 | "log/slog" |
| 9 | "path/filepath" |
| 10 | "testing" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/benbjohnson/litestream" |
| 14 | lss3 "github.com/benbjohnson/litestream/s3" |
| 15 | "github.com/superfly/ltx" |
| 16 | ) |
| 17 | |
| 18 | type generationFake struct { |
| 19 | calls int |
| 20 | ctx context.Context |
| 21 | files []*ltx.FileInfo |
| 22 | } |
| 23 | |
| 24 | func (f *generationFake) Type() string { return "fake" } |
| 25 | func (f *generationFake) SetLogger(*slog.Logger) {} |
| 26 | func (f *generationFake) Init(ctx context.Context) error { f.calls++; f.ctx = ctx; return nil } |
| 27 | func (f *generationFake) LTXFiles(ctx context.Context, _ int, _ ltx.TXID, _ bool) (ltx.FileIterator, error) { |
| 28 | f.calls++ |
| 29 | f.ctx = ctx |
| 30 | return ltx.NewFileInfoSliceIterator(f.files), nil |
| 31 | } |
| 32 | func (f *generationFake) OpenLTXFile(ctx context.Context, _ int, _ ltx.TXID, _ ltx.TXID, _ int64, _ int64) (io.ReadCloser, error) { |
| 33 | f.calls++ |
| 34 | f.ctx = ctx |
| 35 | return io.NopCloser(bytes.NewReader(nil)), nil |
| 36 | } |
| 37 | func (f *generationFake) WriteLTXFile(ctx context.Context, _ int, min, max ltx.TXID, _ io.Reader) (*ltx.FileInfo, error) { |
| 38 | f.calls++ |
| 39 | f.ctx = ctx |
| 40 | return <x.FileInfo{MinTXID: min, MaxTXID: max}, nil |
| 41 | } |
| 42 | func (f *generationFake) DeleteLTXFiles(ctx context.Context, _ []*ltx.FileInfo) error { |
| 43 | f.calls++ |
| 44 | f.ctx = ctx |
| 45 | return nil |
| 46 | } |
| 47 | func (f *generationFake) DeleteAll(context.Context) error { f.calls++; return nil } |
| 48 | |
| 49 | func TestBindGenerationIsImmutableAndCanonical(t *testing.T) { |
| 50 | leaser := lss3.NewLeaser() |
| 51 | leaser.Path = "instances/shop.example/inc" |
| 52 | f := &generationFake{} |
| 53 | h := &Held{leaser: leaser, replicaClient: f} |
| 54 | check := func(context.Context) error { return nil } |
| 55 | if err := h.BindGeneration("instances/shop.example/inc/g-generation1", check); err != nil { |
| 56 | t.Fatal(err) |
| 57 | } |
| 58 | if err := h.BindGeneration("instances/shop.example/inc/g-generation2", check); err != ErrGenerationBinding { |
| 59 | t.Fatalf("rebind=%v", err) |
| 60 | } |
| 61 | for _, bad := range []string{"instances/shop.example/inc/g-a/child", "instances/shop.example/inc/x/../g-a", "instances/shop.example/other/g-a", "instances/shop.example/inc/g-"} { |
| 62 | h2 := &Held{leaser: leaser, replicaClient: &generationFake{}} |
| 63 | if err := h2.BindGeneration(bad, check); err != ErrGenerationBinding { |
| 64 | t.Errorf("BindGeneration(%q)=%v", bad, err) |
| 65 | } |
| 66 | } |
| 67 | } |
| 68 | |
| 69 | func TestGenerationClientValidatesBeforeStorage(t *testing.T) { |
| 70 | f := &generationFake{} |
| 71 | c := &generationClient{client: f, check: func(context.Context) error { return nil }} |
| 72 | if _, err := c.WriteLTXFile(t.Context(), -1, 1, 1, bytes.NewReader(nil)); err != ErrGenerationBinding { |
| 73 | t.Fatal(err) |
| 74 | } |
| 75 | if _, err := c.OpenLTXFile(t.Context(), 0, 2, 1, 0, 0); err != ErrGenerationBinding { |
| 76 | t.Fatal(err) |
| 77 | } |
| 78 | if err := c.DeleteLTXFiles(t.Context(), []*ltx.FileInfo{nil}); err != ErrGenerationBinding { |
| 79 | t.Fatal(err) |
| 80 | } |
| 81 | if err := c.DeleteAll(t.Context()); err == nil { |
| 82 | t.Fatal("DeleteAll accepted") |
| 83 | } |
| 84 | if f.calls != 0 { |
| 85 | t.Fatalf("invalid requests reached storage: %d", f.calls) |
| 86 | } |
| 87 | } |
| 88 | |
| 89 | func TestGenerationClientBoundsAndRetainsStreamContext(t *testing.T) { |
| 90 | checks := 0 |
| 91 | f := &generationFake{} |
| 92 | c := &generationClient{client: f, check: func(context.Context) error { checks++; return nil }} |
| 93 | r, err := c.OpenLTXFile(t.Context(), litestream.SnapshotLevel, 1, 1, 0, 0) |
| 94 | if err != nil { |
| 95 | t.Fatal(err) |
| 96 | } |
| 97 | deadline, ok := f.ctx.Deadline() |
| 98 | if !ok || time.Until(deadline) > generationRequestLimit || time.Until(deadline) < generationRequestLimit-time.Second { |
| 99 | t.Fatalf("deadline=%v ok=%v", deadline, ok) |
| 100 | } |
| 101 | select { |
| 102 | case <-f.ctx.Done(): |
| 103 | t.Fatal("stream context canceled before Close") |
| 104 | default: |
| 105 | } |
| 106 | if err := r.Close(); err != nil { |
| 107 | t.Fatal(err) |
| 108 | } |
| 109 | select { |
| 110 | case <-f.ctx.Done(): |
| 111 | case <-time.After(time.Second): |
| 112 | t.Fatal("stream Close did not cancel request context") |
| 113 | } |
| 114 | if checks != 1 { |
| 115 | t.Fatalf("authority checks=%d", checks) |
| 116 | } |
| 117 | } |
| 118 | |
| 119 | // GenerationTestBinding is a test-only bridge for the external package to |
| 120 | // exercise hostbox.GenerationPrefix without a lease -> hostbox import cycle. |
| 121 | func GenerationTestBinding(location, prefix string) error { |
| 122 | l := lss3.NewLeaser() |
| 123 | l.Path = location |
| 124 | h := &Held{leaser: l, replicaClient: &generationFake{}} |
| 125 | return h.BindGeneration(prefix, func(context.Context) error { return nil }) |
| 126 | } |
| 127 | |
| 128 | func TestGenerationClientRetentionRefusedButUploadsContinue(t *testing.T) { |
| 129 | f := &generationFake{} |
| 130 | c := &generationClient{client: f, check: func(context.Context) error { return nil }} |
| 131 | for _, level := range []int{0, 1, litestream.SnapshotLevel} { |
| 132 | if err := c.DeleteLTXFiles(t.Context(), []*ltx.FileInfo{{Level: level, MinTXID: 1, MaxTXID: 2}}); !errors.Is(err, ErrGenerationRetention) { |
| 133 | t.Fatalf("level %d delete: %v", level, err) |
| 134 | } |
| 135 | } |
| 136 | if f.calls != 0 { |
| 137 | t.Fatal("bare-key deletion reached backing client") |
| 138 | } |
| 139 | if _, err := c.WriteLTXFile(t.Context(), 0, 3, 3, bytes.NewReader(nil)); err != nil { |
| 140 | t.Fatal(err) |
| 141 | } |
| 142 | if f.calls != 1 { |
| 143 | t.Fatal("replication stopped after refusal") |
| 144 | } |
| 145 | } |
| 146 | |
| 147 | func TestGenerationClientDisablesRemoteRetentionInRealStore(t *testing.T) { |
| 148 | for _, bound := range []bool{false, true} { |
| 149 | l := lss3.NewLeaser() |
| 150 | l.Path = "instances/x.example/inc" |
| 151 | h := &Held{leaser: l, replicaClient: &generationFake{}} |
| 152 | if bound { |
| 153 | if err := h.BindGeneration(l.Path+"/g-one/", func(context.Context) error { return nil }); err != nil { |
| 154 | t.Fatal(err) |
| 155 | } |
| 156 | } |
| 157 | _, db, err := h.newStore(filepath.Join(t.TempDir(), "db"), time.Second) |
| 158 | if err != nil { |
| 159 | t.Fatal(err) |
| 160 | } |
| 161 | if db.RetentionEnabled == bound { |
| 162 | t.Fatalf("bound=%v retention=%v", bound, db.RetentionEnabled) |
| 163 | } |
| 164 | } |
| 165 | } |
| 166 | |
| 167 | func TestGenerationClientRealRetentionMonitorsDoNotDelete(t *testing.T) { |
| 168 | l := lss3.NewLeaser() |
| 169 | l.Path = "instances/x.example/inc" |
| 170 | f := &generationFake{files: []*ltx.FileInfo{{Level: 1, MinTXID: 1, MaxTXID: 2, CreatedAt: time.Unix(1, 0)}, {Level: 1, MinTXID: 3, MaxTXID: 4, CreatedAt: time.Unix(2, 0)}}} |
| 171 | h := &Held{leaser: l, replicaClient: f} |
| 172 | if err := h.BindGeneration(l.Path+"/g-one/", func(context.Context) error { return nil }); err != nil { |
| 173 | t.Fatal(err) |
| 174 | } |
| 175 | store, db, err := h.newStore(filepath.Join(t.TempDir(), "db"), time.Second) |
| 176 | if err != nil { |
| 177 | t.Fatal(err) |
| 178 | } |
| 179 | if store.RetentionEnabled || db.RetentionEnabled { |
| 180 | t.Fatal("remote retention enabled") |
| 181 | } |
| 182 | db.MonitorInterval = 0 |
| 183 | db.Replica.MonitorEnabled = false |
| 184 | if err := db.Open(); err != nil { |
| 185 | t.Fatal(err) |
| 186 | } |
| 187 | defer db.Close(t.Context()) |
| 188 | // These execute the real DB and compactor retention code. A deletion |
| 189 | // attempt would get ErrGenerationRetention and fail the call. |
| 190 | if err := db.EnforceRetentionByTXID(t.Context(), 1, 10); err != nil { |
| 191 | t.Fatal(err) |
| 192 | } |
| 193 | if _, err := db.EnforceSnapshotRetention(t.Context(), time.Now()); err != nil { |
| 194 | t.Fatal(err) |
| 195 | } |
| 196 | if err := db.EnforceL0RetentionByTime(t.Context()); err != nil { |
| 197 | t.Fatal(err) |
| 198 | } |
| 199 | if _, err := db.Replica.Client.WriteLTXFile(t.Context(), 0, 5, 5, bytes.NewReader(nil)); err != nil { |
| 200 | t.Fatal(err) |
| 201 | } |
| 202 | } |
| 203 | |
| 204 | // BindRoot gives the root layout the same per-request authority a bound |
| 205 | // generation has (moves Task 5 fix round 1, ruling 2): every replica request |
| 206 | // below the lease's own path first passes check, so an authority withdrawn |
| 207 | // between two requests refuses the second before it reaches storage. It is |
| 208 | // as immutable as BindGeneration, and the two exclude each other. |
| 209 | func TestBindRootChecksEveryRequest(t *testing.T) { |
| 210 | leaser := lss3.NewLeaser() |
| 211 | leaser.Path = "instances/shop.example" |
| 212 | f := &generationFake{} |
| 213 | h := &Held{leaser: leaser, replicaClient: f} |
| 214 | withdrawn := errors.New("the claim moved") |
| 215 | calls := 0 |
| 216 | check := func(context.Context) error { |
| 217 | calls++ |
| 218 | if calls > 1 { |
| 219 | return withdrawn |
| 220 | } |
| 221 | return nil |
| 222 | } |
| 223 | if err := h.BindRoot(check); err != nil { |
| 224 | t.Fatal(err) |
| 225 | } |
| 226 | if err := h.replicaClient.Init(t.Context()); err != nil { |
| 227 | t.Fatalf("first request: %v", err) |
| 228 | } |
| 229 | if _, err := h.replicaClient.LTXFiles(t.Context(), 0, 0, false); !errors.Is(err, withdrawn) { |
| 230 | t.Fatalf("second request after the authority was withdrawn: %v", err) |
| 231 | } |
| 232 | if f.calls != 1 { |
| 233 | t.Fatalf("storage saw %d requests, want the first only", f.calls) |
| 234 | } |
| 235 | if err := h.BindRoot(check); err != ErrGenerationBinding { |
| 236 | t.Fatalf("rebind=%v", err) |
| 237 | } |
| 238 | if err := h.BindGeneration("instances/shop.example/g-a", check); err != ErrGenerationBinding { |
| 239 | t.Fatalf("a generation over the root binding=%v", err) |
| 240 | } |
| 241 | h2 := &Held{leaser: leaser, replicaClient: &generationFake{}} |
| 242 | if err := h2.BindGeneration("instances/shop.example/g-a", check); err != nil { |
| 243 | t.Fatal(err) |
| 244 | } |
| 245 | if err := h2.BindRoot(check); err != ErrGenerationBinding { |
| 246 | t.Fatalf("the root over a generation binding=%v", err) |
| 247 | } |
| 248 | if err := (&Held{leaser: leaser}).BindRoot(nil); err != ErrGenerationBinding { |
| 249 | t.Fatalf("nil check=%v", err) |
| 250 | } |
| 251 | } |
| 252 | |
amadan