| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "bufio" |
| 5 | "context" |
| 6 | "crypto/sha256" |
| 7 | "encoding/hex" |
| 8 | "encoding/json" |
| 9 | "net/http" |
| 10 | "os" |
| 11 | "strconv" |
| 12 | "strings" |
| 13 | "sync" |
| 14 | "testing" |
| 15 | "time" |
| 16 | |
| 17 | "github.com/aws/aws-sdk-go-v2/aws" |
| 18 | "github.com/aws/aws-sdk-go-v2/credentials" |
| 19 | awss3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| 20 | "github.com/benbjohnson/litestream" |
| 21 | lsfile "github.com/benbjohnson/litestream/file" |
| 22 | lss3 "github.com/benbjohnson/litestream/s3" |
| 23 | |
| 24 | "github.com/carlosframework/platform/internal/dbdoor" |
| 25 | "github.com/carlosframework/platform/internal/dbdoor/dbdoortest" |
| 26 | "github.com/carlosframework/platform/internal/dbreads" |
| 27 | "github.com/carlosframework/platform/internal/dbreads/dbreadstest" |
| 28 | "github.com/carlosframework/platform/internal/provision" |
| 29 | "github.com/carlosframework/platform/internal/release" |
| 30 | ) |
| 31 | |
| 32 | // uploadReplica writes a file replica's LTX files into an S3 replica the way |
| 33 | // a box's litestream does (WriteLTXFile), level by level. |
| 34 | func uploadReplica(t *testing.T, from *lsfile.ReplicaClient, to *lss3.ReplicaClient) { |
| 35 | t.Helper() |
| 36 | ctx := context.Background() |
| 37 | for level := 0; level <= litestream.SnapshotLevel; level++ { |
| 38 | itr, err := from.LTXFiles(ctx, level, 0, false) |
| 39 | if err != nil { |
| 40 | t.Fatal(err) |
| 41 | } |
| 42 | for itr.Next() { |
| 43 | f := itr.Item() |
| 44 | rc, err := from.OpenLTXFile(ctx, level, f.MinTXID, f.MaxTXID, 0, 0) |
| 45 | if err != nil { |
| 46 | t.Fatal(err) |
| 47 | } |
| 48 | if _, err := to.WriteLTXFile(ctx, level, f.MinTXID, f.MaxTXID, rc); err != nil { |
| 49 | t.Fatal(err) |
| 50 | } |
| 51 | rc.Close() |
| 52 | } |
| 53 | itr.Close() |
| 54 | } |
| 55 | } |
| 56 | |
| 57 | // dbfGateStore holds the box's chunk and result creates, once armed, until |
| 58 | // release closes, and records when each landed: T16 needs a box that |
| 59 | // publishes only after the console has ended the request (plan review |
| 60 | // finding 19). |
| 61 | type dbfGateStore struct { |
| 62 | release.Store |
| 63 | mu sync.Mutex |
| 64 | open chan struct{} |
| 65 | reached chan struct{} |
| 66 | once *sync.Once |
| 67 | landed []time.Time |
| 68 | } |
| 69 | |
| 70 | func (g *dbfGateStore) arm() { |
| 71 | g.mu.Lock() |
| 72 | defer g.mu.Unlock() |
| 73 | g.open, g.reached, g.once, g.landed = make(chan struct{}), make(chan struct{}), &sync.Once{}, nil |
| 74 | } |
| 75 | |
| 76 | func (g *dbfGateStore) release() { |
| 77 | g.mu.Lock() |
| 78 | defer g.mu.Unlock() |
| 79 | close(g.open) |
| 80 | } |
| 81 | |
| 82 | func (g *dbfGateStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| 83 | g.mu.Lock() |
| 84 | open, reached, once := g.open, g.reached, g.once |
| 85 | g.mu.Unlock() |
| 86 | if open == nil || !(strings.Contains(key, "/out/") || strings.HasSuffix(key, "/result.json")) { |
| 87 | return g.Store.Put(ctx, key, body, ifMatch) |
| 88 | } |
| 89 | once.Do(func() { close(reached) }) |
| 90 | <-open |
| 91 | etag, err := g.Store.Put(ctx, key, body, ifMatch) |
| 92 | if err == nil { |
| 93 | g.mu.Lock() |
| 94 | g.landed = append(g.landed, time.Now()) |
| 95 | g.mu.Unlock() |
| 96 | } |
| 97 | return etag, err |
| 98 | } |
| 99 | |
| 100 | func (g *dbfGateStore) landedAfter(t time.Time) (n int, early bool) { |
| 101 | g.mu.Lock() |
| 102 | defer g.mu.Unlock() |
| 103 | for _, at := range g.landed { |
| 104 | if !at.After(t) { |
| 105 | early = true |
| 106 | } |
| 107 | } |
| 108 | return len(g.landed), early |
| 109 | } |
| 110 | |
| 111 | // TestDBForwardMinio is T1 and T16 against a real S3 server: a console in ie |
| 112 | // and a box in uk sharing one deployment bucket (versioned), each with its |
| 113 | // own data bucket and its own credentials, on one endpoint (spec §12). The |
| 114 | // console's credentials cannot read uk's bucket (a MinIO policy), so a |
| 115 | // console that read it would fail. Two tenants in uk each get their own |
| 116 | // rows. Neither the console nor the box ever deletes or replaces anything |
| 117 | // under db-reads/, and a box's writes after the request ended change |
| 118 | // nothing. It runs only where .amadan/ci.d/30-minio-contract stands MinIO |
| 119 | // up; the box is plan 1's real code joined in process (dbreadstest). |
| 120 | func TestDBForwardMinio(t *testing.T) { |
| 121 | if os.Getenv("CARLOS_MINIO_TEST") != "1" { |
| 122 | t.Skip("set CARLOS_MINIO_TEST=1 (with a running MinIO and the DBF_* variables) to run; see .amadan/ci.d/30-minio-contract") |
| 123 | } |
| 124 | need := func(name string) string { |
| 125 | v := os.Getenv(name) |
| 126 | if v == "" { |
| 127 | t.Fatalf("%s is not set", name) |
| 128 | } |
| 129 | return v |
| 130 | } |
| 131 | endpoint := need("MINIO_ENDPOINT") |
| 132 | rootKey, rootSecret := need("DBF_ROOT_KEY"), need("DBF_ROOT_SECRET") |
| 133 | consoleKey, consoleSecret := need("DBF_CONSOLE_KEY"), need("DBF_CONSOLE_SECRET") |
| 134 | boxKey, boxSecret := need("DBF_BOX_KEY"), need("DBF_BOX_SECRET") |
| 135 | store := func(key, secret, bucket string) *release.S3Store { |
| 136 | return &release.S3Store{Bucket: bucket, Client: awss3.New(awss3.Options{ |
| 137 | Region: "us-east-1", Credentials: credentials.NewStaticCredentialsProvider(key, secret, ""), |
| 138 | BaseEndpoint: aws.String(endpoint), UsePathStyle: true, |
| 139 | RequestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired, |
| 140 | ResponseChecksumValidation: aws.ResponseChecksumValidationWhenRequired, |
| 141 | })} |
| 142 | } |
| 143 | replica := func(key, secret, bucket, prefix string) *lss3.ReplicaClient { |
| 144 | c := lss3.NewReplicaClient() |
| 145 | c.AccessKeyID, c.SecretAccessKey, c.Region, c.Bucket, c.Path, c.Endpoint, c.ForcePathStyle = |
| 146 | key, secret, "us-east-1", bucket, prefix, endpoint, true |
| 147 | return c |
| 148 | } |
| 149 | run := strconv.FormatInt(time.Now().UnixNano(), 36) |
| 150 | shop, other := "shop-"+run+".example.com", "other-"+run+".example.com" |
| 151 | for host, batches := range map[string]int{shop: 3, other: 1} { |
| 152 | dir := t.TempDir() |
| 153 | dbdoortest.Replicate(t, dir, batches) |
| 154 | uploadReplica(t, lsfile.NewReplicaClient(dir), replica(rootKey, rootSecret, "dbf-data-uk", "instances/"+host)) |
| 155 | } |
| 156 | |
| 157 | // The policy, shown: the console's credentials cannot list uk's bucket. |
| 158 | itr, err := replica(consoleKey, consoleSecret, "dbf-data-uk", "instances/"+shop).LTXFiles(context.Background(), 0, 0, false) |
| 159 | if err == nil { |
| 160 | for itr.Next() { |
| 161 | } |
| 162 | err = itr.Err() |
| 163 | itr.Close() |
| 164 | } |
| 165 | if err == nil { |
| 166 | t.Fatal("the console's credentials read uk's data bucket: the test proves nothing") |
| 167 | } |
| 168 | |
| 169 | deploy := store(consoleKey, consoleSecret, "dbf-deploy") |
| 170 | fx := newConsoleFixtureOn(t, true, strictStore{Store: deploy, t: t}) |
| 171 | fx.srv.cfg.DBForward, fx.srv.cfg.Region, fx.srv.cfg.HibernateBucket = true, "ie", "dbf-data-ie" |
| 172 | fx.srv.dbReplica = func(_ context.Context, prefix string) (replicaReader, error) { |
| 173 | return replica(consoleKey, consoleSecret, "dbf-data-ie", prefix), nil // the only data bucket this console has |
| 174 | } |
| 175 | if _, err := fx.st.CreateAppIn(OpsAccountID, "shop", "paul@keymail.dev", time.Now()); err != nil { |
| 176 | t.Fatal(err) |
| 177 | } |
| 178 | if err := fx.st.AddMember(OpsAccountID, "member@example.com", RoleMember, time.Now()); err != nil { |
| 179 | t.Fatal(err) |
| 180 | } |
| 181 | member := cliToken(t, fx, "member@example.com", "code-minio") |
| 182 | for _, host := range []string{shop, other} { |
| 183 | putShopRecord(t, fx, provision.Record{Host: host, Region: "uk", CreatedAt: time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)}) |
| 184 | } |
| 185 | putEdgesFile(t, fx.rel, "uk-"+run, "uk", []string{dbreads.Capability}, false) |
| 186 | putBoxStatus(t, fx.rel, "uk-"+run, time.Now()) |
| 187 | gate := &dbfGateStore{Store: store(boxKey, boxSecret, "dbf-deploy")} |
| 188 | box := dbreadstest.Start(t, dbreadstest.Options{ |
| 189 | Store: strictStore{Store: gate, t: t}, Region: "uk", Box: "uk-" + run, |
| 190 | Pub: fx.pub, HibernatePrefix: "instances", |
| 191 | OpenReplica: func(_ context.Context, prefix string) (dbdoor.ReplicaReader, error) { |
| 192 | return replica(boxKey, boxSecret, "dbf-data-uk", prefix), nil |
| 193 | }, |
| 194 | }) |
| 195 | runBox(t, box) |
| 196 | dfx := &dbfFixture{consoleFixture: fx, member: member} |
| 197 | |
| 198 | // T1: a forwarded query end to end, for each tenant, each its own rows. |
| 199 | for host, want := range map[string]string{shop: "n\n30\n", other: "n\n10\n"} { |
| 200 | body, key := queryBody(t, "SELECT count(*) AS n FROM t") |
| 201 | res := dfx.openRead(t, host, member, body) |
| 202 | lines := readLines(t, res.Body) |
| 203 | res.Body.Close() |
| 204 | if rows, _ := openAnswer(t, key, lines); string(rows) != want { |
| 205 | t.Fatalf("%s: %q, want %q", host, rows, want) |
| 206 | } |
| 207 | } |
| 208 | |
| 209 | // A pull through MinIO (plan 3): the planned files, sealed as the ltx |
| 210 | // stream, adding up to the summary, whose file list ends at its copy. |
| 211 | pkey, _ := dbreads.NewReplyKey() |
| 212 | pres := dfx.openRead(t, shop, member, dbreads.ReadBody{Op: dbreads.OpPull, ReplyKey: dbreads.EncodeReplyKey(pkey.PublicKey())}) |
| 213 | plines := readLines(t, pres.Body) |
| 214 | pres.Body.Close() |
| 215 | presult := plines[len(plines)-1].Result |
| 216 | if presult == nil || presult.State != dbreads.StateDone { |
| 217 | t.Fatalf("pull: %+v", plines[len(plines)-1]) |
| 218 | } |
| 219 | psum, err := dbreads.OpenSummary(pkey, plines[0].Admitted, presult.Sealed) |
| 220 | if err != nil || len(psum.Files) == 0 || psum.TXID != psum.Files[len(psum.Files)-1].MaxTXID { |
| 221 | t.Fatalf("pull summary %+v %v", psum, err) |
| 222 | } |
| 223 | ph := sha256.New() |
| 224 | var pn int64 |
| 225 | for _, l := range plines { |
| 226 | if l.Chunk == 0 { |
| 227 | continue |
| 228 | } |
| 229 | plain, err := dbreads.OpenChunk(pkey, plines[0].Admitted, l.Chunk, dbreads.StreamLTX, l.Chunk == presult.LastSeq, l.Data) |
| 230 | if err != nil { |
| 231 | t.Fatalf("pull chunk %d: %v", l.Chunk, err) |
| 232 | } |
| 233 | ph.Write(plain) |
| 234 | pn += int64(len(plain)) |
| 235 | } |
| 236 | if hex.EncodeToString(ph.Sum(nil)) != psum.SHA256 || pn != psum.Bytes { |
| 237 | t.Fatal("the pulled bytes do not add up to the summary") |
| 238 | } |
| 239 | |
| 240 | // T16 (plan review finding 19): the real box reaches publication and is |
| 241 | // held there; the console's request ends, by disconnect and then by its |
| 242 | // bound; only then does the box publish. Its late chunks and result land |
| 243 | // after the console ended, change nothing the console sends, and the |
| 244 | // strict stores see no delete and no replacing write from either side. |
| 245 | late := func(t *testing.T, end func(res *http.Response, br *bufio.Reader)) { |
| 246 | gate.arm() |
| 247 | body, _ := queryBody(t, "SELECT count(*) AS n FROM t") |
| 248 | res := dfx.openRead(t, shop, member, body) |
| 249 | br := bufio.NewReader(res.Body) |
| 250 | line, _ := br.ReadBytes('\n') |
| 251 | var adm dbreads.Line |
| 252 | json.Unmarshal(line, &adm) |
| 253 | end(res, br) |
| 254 | // The handler's last act is the cancel: once it is there, the |
| 255 | // console has ended the request. |
| 256 | waitKey := func(key string) { |
| 257 | deadline := time.Now().Add(60 * time.Second) |
| 258 | for { |
| 259 | if _, _, err := deploy.Get(context.Background(), key); err == nil { |
| 260 | return |
| 261 | } |
| 262 | if time.Now().After(deadline) { |
| 263 | t.Fatalf("no %s", key) |
| 264 | } |
| 265 | time.Sleep(100 * time.Millisecond) |
| 266 | } |
| 267 | } |
| 268 | waitKey(dbreads.CancelKey(adm.Admitted)) |
| 269 | consoleEnded := time.Now() |
| 270 | select { |
| 271 | case <-gate.reached: |
| 272 | case <-time.After(60 * time.Second): |
| 273 | t.Fatal("the box never reached publication") |
| 274 | } |
| 275 | if keys, _ := deploy.List(context.Background(), dbreads.OutPrefix(adm.Admitted)); len(keys) != 0 { |
| 276 | t.Fatalf("output published before the console ended: %v", keys) |
| 277 | } |
| 278 | gate.release() |
| 279 | waitKey(dbreads.ResultKey(adm.Admitted)) |
| 280 | // At least the result lands late; a held chunk lands too unless the |
| 281 | // owner's cancel watch ended its create first (plan 1's watchCancel). |
| 282 | if n, early := gate.landedAfter(consoleEnded); n < 1 || early { |
| 283 | t.Fatalf("%d late writes, any before the console ended: %v", n, early) |
| 284 | } |
| 285 | } |
| 286 | t.Run("by disconnect", func(t *testing.T) { |
| 287 | late(t, func(res *http.Response, _ *bufio.Reader) { |
| 288 | select { |
| 289 | case <-gate.reached: |
| 290 | case <-time.After(60 * time.Second): |
| 291 | t.Fatal("the box never reached publication") |
| 292 | } |
| 293 | res.Body.Close() |
| 294 | }) |
| 295 | }) |
| 296 | t.Run("by timeout", func(t *testing.T) { |
| 297 | // The request's own bound, well past the box's restore and query of |
| 298 | // three batches, well short of the box's 20-minute owner deadline. |
| 299 | fx.srv.dbf.bound, fx.srv.dbf.grace = 20*time.Second, 0 |
| 300 | late(t, func(res *http.Response, br *bufio.Reader) { |
| 301 | defer res.Body.Close() |
| 302 | lines := readLines(t, br) |
| 303 | if last := lines[len(lines)-1]; last.Ended != dbreads.EndedNoResult { |
| 304 | t.Fatalf("by timeout: %+v", last) |
| 305 | } |
| 306 | for _, l := range lines { |
| 307 | if l.Chunk != 0 || l.Result != nil { |
| 308 | t.Fatalf("a held box's output reached the console: %+v", l) |
| 309 | } |
| 310 | } |
| 311 | }) |
| 312 | }) |
| 313 | } |
| 314 | |
amadan