Test backfill of late arriving sequences to the channel caches
(t *testing.T)
| 503 | |
| 504 | // Test backfill of late arriving sequences to the channel caches |
| 505 | func TestChannelCacheBackfill(t *testing.T) { |
| 506 | |
| 507 | base.SetUpTestLogging(t, base.LevelDebug, base.KeyCache, base.KeyChanges) |
| 508 | |
| 509 | db, ctx := setupTestDBWithCacheOptions(t, shortWaitCache()) |
| 510 | defer db.Close(ctx) |
| 511 | |
| 512 | collection := GetSingleDatabaseCollection(t, db.DatabaseContext) |
| 513 | // Create a user with access to channel ABC |
| 514 | authenticator := db.Authenticator(ctx) |
| 515 | user, err := authenticator.NewUser("naomi", "letmein", channels.BaseSetOf(t, "ABC", "PBS", "NBC", "TBS")) |
| 516 | require.NoError(t, err) |
| 517 | require.NoError(t, authenticator.Save(user)) |
| 518 | |
| 519 | // Simulate seq 3 being delayed - write 1,2,4,5 |
| 520 | WriteDirect(t, collection, []string{"ABC", "NBC"}, 1) |
| 521 | WriteDirect(t, collection, []string{"ABC"}, 2) |
| 522 | WriteDirect(t, collection, []string{"ABC", "PBS"}, 5) |
| 523 | WriteDirect(t, collection, []string{"ABC", "PBS"}, 6) |
| 524 | |
| 525 | // Test that retrieval isn't blocked by skipped sequences |
| 526 | require.NoError(t, db.changeCache.waitForSequence(ctx, 6, base.DefaultWaitForSequence)) |
| 527 | collectionWithUser, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db) |
| 528 | collectionWithUser.user, err = authenticator.GetUser("naomi") |
| 529 | require.NoError(t, err) |
| 530 | changes := getChanges(t, collectionWithUser, base.SetOf("*"), getChangesOptionsWithZeroSeq(t)) |
| 531 | |
| 532 | collectionID := collection.GetCollectionID() |
| 533 | |
| 534 | assert.Equal(t, &ChangeEntry{ |
| 535 | Seq: SequenceID{Seq: 1, TriggeredBy: 0, LowSeq: 2}, |
| 536 | ID: "doc-1", |
| 537 | Changes: []ChangeByVersionType{{"rev": "1-a"}}, |
| 538 | collectionID: collectionID}, changes[0]) |
| 539 | |
| 540 | lastSeq := changes[len(changes)-1].Seq |
| 541 | |
| 542 | // Validate insert to various cache states |
| 543 | WriteDirect(t, collection, []string{"ABC", "NBC", "PBS", "TBS"}, 3) |
| 544 | WriteDirect(t, collection, []string{"CBS"}, 7) |
| 545 | require.NoError(t, db.changeCache.waitForSequence(ctx, 7, base.DefaultWaitForSequence)) |
| 546 | |
| 547 | // verify insert at start (PBS) |
| 548 | pbsCache, err := db.changeCache.getChannelCache().getSingleChannelCache(ctx, channels.NewID("PBS", collectionID)) |
| 549 | require.NoError(t, err) |
| 550 | assert.True(t, verifyCacheSequences(pbsCache, []uint64{3, 5, 6})) |
| 551 | // verify insert at middle (ABC) |
| 552 | abcCache, err := db.changeCache.getChannelCache().getSingleChannelCache(ctx, channels.NewID("ABC", collectionID)) |
| 553 | require.NoError(t, err) |
| 554 | assert.True(t, verifyCacheSequences(abcCache, []uint64{1, 2, 3, 5, 6})) |
| 555 | // verify insert at end (NBC) |
| 556 | nbcCache, err := db.changeCache.getChannelCache().getSingleChannelCache(ctx, channels.NewID("NBC", collectionID)) |
| 557 | require.NoError(t, err) |
| 558 | assert.True(t, verifyCacheSequences(nbcCache, []uint64{1, 3})) |
| 559 | // verify insert to empty cache (TBS) |
| 560 | tbsCache, err := db.changeCache.getChannelCache().getSingleChannelCache(ctx, channels.NewID("TBS", collectionID)) |
| 561 | require.NoError(t, err) |
| 562 | assert.True(t, verifyCacheSequences(tbsCache, []uint64{3})) |
nothing calls this directly
no test coverage detected