| 1 | package release |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "log" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "strings" |
| 12 | "sync/atomic" |
| 13 | "testing" |
| 14 | ) |
| 15 | |
| 16 | const bumpTestKey = "apps/bab/shop/instances/h.json" |
| 17 | const bumpTestScope = "apps/bab/shop" |
| 18 | |
| 19 | func indexedDirStore(t *testing.T) *DirStore { |
| 20 | t.Helper() |
| 21 | d := NewDirStore(t.TempDir()) |
| 22 | d.EnableChangeIndex() |
| 23 | return d |
| 24 | } |
| 25 | |
| 26 | func scopeGen(t *testing.T, s Store, scope string) uint64 { |
| 27 | t.Helper() |
| 28 | idx, _, err := ReadChangeIndex(context.Background(), s) |
| 29 | if err != nil { |
| 30 | t.Fatal(err) |
| 31 | } |
| 32 | return idx.Entries[scope] |
| 33 | } |
| 34 | |
| 35 | func TestStoreBumpsAfterEveryMutation(t *testing.T) { |
| 36 | t.Parallel() |
| 37 | ctx := context.Background() |
| 38 | d := indexedDirStore(t) |
| 39 | want := uint64(0) |
| 40 | step := func(name string, err error) { |
| 41 | t.Helper() |
| 42 | if err != nil { |
| 43 | t.Fatalf("%s: %v", name, err) |
| 44 | } |
| 45 | want++ |
| 46 | if got := scopeGen(t, d, bumpTestScope); got != want { |
| 47 | t.Fatalf("after %s: gen = %d, want %d", name, got, want) |
| 48 | } |
| 49 | } |
| 50 | etag, err := d.Put(ctx, bumpTestKey, []byte("one"), "") |
| 51 | step("Put", err) |
| 52 | etag, err = d.PutFrom(ctx, bumpTestKey, bytes.NewReader([]byte("two")), 3, etag) |
| 53 | step("PutFrom", err) |
| 54 | step("DeleteIfMatch", DeleteIfMatch(ctx, d, bumpTestKey, etag)) |
| 55 | _, err = d.Put(ctx, bumpTestKey, []byte("three"), "") |
| 56 | step("Put again", err) |
| 57 | step("Delete", d.Delete(ctx, bumpTestKey)) |
| 58 | } |
| 59 | |
| 60 | func TestStoreDoesNotBumpFailedMutation(t *testing.T) { |
| 61 | t.Parallel() |
| 62 | ctx := context.Background() |
| 63 | d := indexedDirStore(t) |
| 64 | if _, err := d.Put(ctx, bumpTestKey, []byte("one"), ""); err != nil { |
| 65 | t.Fatal(err) |
| 66 | } |
| 67 | if _, err := d.Put(ctx, bumpTestKey, []byte("two"), "stale"); !errors.Is(err, ErrCASConflict) { |
| 68 | t.Fatalf("stale Put = %v, want ErrCASConflict", err) |
| 69 | } |
| 70 | if _, err := d.PutFrom(ctx, bumpTestKey, bytes.NewReader([]byte("two")), 3, "stale"); !errors.Is(err, ErrCASConflict) { |
| 71 | t.Fatalf("stale PutFrom = %v, want ErrCASConflict", err) |
| 72 | } |
| 73 | if err := DeleteIfMatch(ctx, d, bumpTestKey, "stale"); !errors.Is(err, ErrCASConflict) { |
| 74 | t.Fatalf("stale DeleteIfMatch = %v, want ErrCASConflict", err) |
| 75 | } |
| 76 | if got := scopeGen(t, d, bumpTestScope); got != 1 { |
| 77 | t.Fatalf("gen = %d after one success and three refusals, want 1", got) |
| 78 | } |
| 79 | } |
| 80 | |
| 81 | func TestStoreDoesNotBumpUnscopedKeys(t *testing.T) { |
| 82 | t.Parallel() |
| 83 | ctx := context.Background() |
| 84 | d := indexedDirStore(t) |
| 85 | for _, k := range []string{"apps/bab/shop/logs/box/x", "control/serial", "status/boxes/a.json", "releases/x"} { |
| 86 | if _, err := d.Put(ctx, k, []byte("x"), ""); err != nil { |
| 87 | t.Fatal(err) |
| 88 | } |
| 89 | if err := d.Delete(ctx, k); err != nil { |
| 90 | t.Fatal(err) |
| 91 | } |
| 92 | } |
| 93 | if _, _, err := d.Get(ctx, ChangeIndexKey); !errors.Is(err, os.ErrNotExist) { |
| 94 | t.Fatalf("index object after unscoped writes: err = %v, want not-exist", err) |
| 95 | } |
| 96 | // Writing the index itself never recurses or bumps. |
| 97 | if _, err := d.Put(ctx, ChangeIndexKey, []byte(`{"schema":1,"entries":{}}`), ""); err != nil { |
| 98 | t.Fatal(err) |
| 99 | } |
| 100 | if got := scopeGen(t, d, bumpTestScope); got != 0 { |
| 101 | t.Fatalf("gen = %d after writing the index key, want 0", got) |
| 102 | } |
| 103 | } |
| 104 | |
| 105 | func TestStoreBumpFailureDoesNotFailWrite(t *testing.T) { |
| 106 | ctx := context.Background() |
| 107 | d := indexedDirStore(t) |
| 108 | // A directory where the index object belongs: reading it fails with |
| 109 | // something other than not-exist, so every bump attempt errors. |
| 110 | if err := os.MkdirAll(filepath.Join(d.root, filepath.FromSlash(ChangeIndexKey)), 0o755); err != nil { |
| 111 | t.Fatal(err) |
| 112 | } |
| 113 | var logs bytes.Buffer |
| 114 | logTo(t, &logs) |
| 115 | if _, err := d.Put(ctx, bumpTestKey, []byte("one"), ""); err != nil { |
| 116 | t.Fatalf("Put failed because the bump did: %v", err) |
| 117 | } |
| 118 | body, _, err := d.Get(ctx, bumpTestKey) |
| 119 | if err != nil || string(body) != "one" { |
| 120 | t.Fatalf("record = %q, %v; want written", body, err) |
| 121 | } |
| 122 | if !strings.Contains(logs.String(), "release: change index bump "+bumpTestScope+":") { |
| 123 | t.Fatalf("bump failure was not logged: %q", logs.String()) |
| 124 | } |
| 125 | } |
| 126 | |
| 127 | func TestStoreBumpDetachedFromCallerCancel(t *testing.T) { |
| 128 | t.Parallel() |
| 129 | d := indexedDirStore(t) |
| 130 | ctx, cancel := context.WithCancel(context.Background()) |
| 131 | var seenCancelled atomic.Bool |
| 132 | get := func(c context.Context, k string) ([]byte, string, error) { |
| 133 | if c.Err() != nil { |
| 134 | seenCancelled.Store(true) |
| 135 | } |
| 136 | return d.Get(c, k) |
| 137 | } |
| 138 | put := func(c context.Context, k string, b []byte, m string) (string, error) { |
| 139 | if c.Err() != nil { |
| 140 | seenCancelled.Store(true) |
| 141 | } |
| 142 | return d.putRaw(c, k, b, m) |
| 143 | } |
| 144 | // The caller's request ends the moment its write returns. |
| 145 | cancel() |
| 146 | bumpAfterMutation(ctx, true, bumpTestKey, get, put) |
| 147 | if seenCancelled.Load() { |
| 148 | t.Fatal("the bump ran under the caller's cancelled context") |
| 149 | } |
| 150 | if got := scopeGen(t, d, bumpTestScope); got != 1 { |
| 151 | t.Fatalf("gen = %d after a bump with a cancelled caller, want 1", got) |
| 152 | } |
| 153 | // And through the store: a cancelled context still writes (DirStore does |
| 154 | // not look at it) and still bumps. |
| 155 | cctx, ccancel := context.WithCancel(context.Background()) |
| 156 | ccancel() |
| 157 | if _, err := d.Put(cctx, bumpTestKey, []byte("x"), ""); err != nil { |
| 158 | t.Fatal(err) |
| 159 | } |
| 160 | if got := scopeGen(t, d, bumpTestScope); got != 2 { |
| 161 | t.Fatalf("gen = %d, want 2", got) |
| 162 | } |
| 163 | } |
| 164 | |
| 165 | func TestStoreWithoutEnableNeverBumps(t *testing.T) { |
| 166 | t.Parallel() |
| 167 | ctx := context.Background() |
| 168 | d := NewDirStore(t.TempDir()) |
| 169 | etag, err := d.Put(ctx, bumpTestKey, []byte("one"), "") |
| 170 | if err != nil { |
| 171 | t.Fatal(err) |
| 172 | } |
| 173 | if _, err := d.PutFrom(ctx, bumpTestKey, bytes.NewReader([]byte("two")), 3, etag); err != nil { |
| 174 | t.Fatal(err) |
| 175 | } |
| 176 | if err := d.Delete(ctx, bumpTestKey); err != nil { |
| 177 | t.Fatal(err) |
| 178 | } |
| 179 | if _, _, err := d.Get(ctx, ChangeIndexKey); !errors.Is(err, os.ErrNotExist) { |
| 180 | t.Fatalf("index object on a store that never enabled it: err = %v", err) |
| 181 | } |
| 182 | } |
| 183 | |
| 184 | func TestStoreCapabilitiesPreserved(t *testing.T) { |
| 185 | t.Parallel() |
| 186 | d := NewDirStore(t.TempDir()) |
| 187 | d.EnableChangeIndex() |
| 188 | var s Store = d |
| 189 | _, ok1 := s.(ConditionalDeleter) |
| 190 | _, ok2 := s.(PutStreamer) |
| 191 | _, ok3 := s.(Opener) |
| 192 | _, ok4 := s.(SizeLister) |
| 193 | _, ok5 := s.(PageLister) |
| 194 | _, ok6 := s.(ModLister) |
| 195 | if !(ok1 && ok2 && ok3 && ok4 && ok5 && ok6) { |
| 196 | t.Fatal("index-enabled DirStore lost an extension") |
| 197 | } |
| 198 | var s3 Store = &S3Store{} |
| 199 | _, ok1 = s3.(ConditionalDeleter) |
| 200 | _, ok2 = s3.(PutStreamer) |
| 201 | _, ok3 = s3.(Opener) |
| 202 | _, ok4 = s3.(SizeLister) |
| 203 | _, ok5 = s3.(PageLister) |
| 204 | _, ok6 = s3.(ModLister) |
| 205 | if !(ok1 && ok2 && ok3 && ok4 && ok5 && ok6) { |
| 206 | t.Fatal("S3Store lost an extension") |
| 207 | } |
| 208 | } |
| 209 | |
| 210 | // TestPurgeBumpsBounded: a purge deletes many keys under one app. Each delete |
| 211 | // is its own mutation and costs exactly one index CAS, so the generation lands |
| 212 | // on the delete count and the write amplification is one put per delete. |
| 213 | func TestPurgeBumpsBounded(t *testing.T) { |
| 214 | t.Parallel() |
| 215 | ctx := context.Background() |
| 216 | d := indexedDirStore(t) |
| 217 | const n = 300 |
| 218 | for i := 0; i < n; i++ { |
| 219 | if err := d.Delete(ctx, fmt.Sprintf("apps/bab/shop/versions/%03d.json", i)); err != nil { |
| 220 | t.Fatal(err) |
| 221 | } |
| 222 | } |
| 223 | if got := scopeGen(t, d, bumpTestScope); got != n { |
| 224 | t.Fatalf("gen = %d after %d deletes, want %d", got, n, n) |
| 225 | } |
| 226 | // One bump is one CAS: count the index puts the hook issues over the raw backend. |
| 227 | var puts atomic.Int64 |
| 228 | put := func(c context.Context, k string, b []byte, m string) (string, error) { |
| 229 | puts.Add(1) |
| 230 | return d.putRaw(c, k, b, m) |
| 231 | } |
| 232 | for i := 0; i < 20; i++ { |
| 233 | bumpAfterMutation(ctx, true, bumpTestKey, d.Get, put) |
| 234 | } |
| 235 | if got := puts.Load(); got != 20 { |
| 236 | t.Fatalf("20 bumps issued %d index puts, want exactly one each", got) |
| 237 | } |
| 238 | } |
| 239 | |
| 240 | // logTo captures the standard logger for one serial test. |
| 241 | func logTo(t *testing.T, w *bytes.Buffer) { |
| 242 | t.Helper() |
| 243 | log.SetOutput(w) |
| 244 | t.Cleanup(func() { log.SetOutput(os.Stderr) }) |
| 245 | } |
| 246 | |
amadan