Function TestResyncUsingDCPStream

rest/adminapitest/admin_api_test.go:687–752

(t *testing.T)

}

func TestResyncUsingDCPStream(t *testing.T) {
	base.TestRequiresDCPResync(t)
	base.LongRunningTest(t)

	testCases := []struct {
		docsCreated int
	}{
		{
			docsCreated: 0,
		},
		{
			docsCreated: 1000,
		},
	}

	syncFn := `
	function(doc) {
		channel("x")
	}`

	for _, testCase := range testCases {
		t.Run(fmt.Sprintf("Docs %d", testCase.docsCreated), func(t *testing.T) {
			rt := rest.NewRestTester(t,
				&rest.RestTesterConfig{
					SyncFn: syncFn,
				},
			)
			defer rt.Close()

			for i := 0; i < testCase.docsCreated; i++ {
				rt.CreateTestDoc(fmt.Sprintf("doc%d", i))
			}

			err := rt.WaitForCondition(func() bool {
				return int(rt.GetDatabase().DbStats.Database().SyncFunctionCount.Value()) == testCase.docsCreated
			})
			assert.NoError(t, err)
			rt.GetDatabase().DbStats.Database().SyncFunctionCount.Set(0)

			response := rt.SendAdminRequest("POST", "/db/_resync?action=start", "")
			rest.RequireStatus(t, response, http.StatusServiceUnavailable)

			rt.TakeDbOffline()

			response = rt.SendAdminRequest("POST", "/db/_resync?action=start", "")
			rest.RequireStatus(t, response, http.StatusOK)

			resyncManagerStatus := rt.WaitForResyncDCPStatus(db.BackgroundProcessStateCompleted)

			assert.Equal(t, testCase.docsCreated, int(rt.GetDatabase().DbStats.Database().SyncFunctionCount.Value()))
			if !base.UnitTestUrlIsWalrus() && !base.TestsDisableGSI() {
				// It is possible for Couchbase Server GSI runs which use DCP purge to two DCP events from a previous
				// test.
				// 1. doc1 mutation
				// 2. doc1 deletion
				//
				// In a test, these will not be resynced but docsProcessed is incremented. Relax
				// the assertion to greater than the number of documents.