compactor: fix TestPeriodic

Perviously, we advance checkCompactionInterval more than we should.
The compaction might happen nondeterministically since there is no
synchronization before we call clock.Advance().

The number of rg.Wait() should be equal to the number of Advance() if
compactor routine and test routine run at the same pace. However, in our current
test, we call Advance() more than rg.Wait().

It works OK when the compactor routine runs "slower" than the test routine, which
is the common case. However, when the speed changes, the compactor routine might
block rg.Rev() since there is not enough calls of rg.Wait().

This commit forces the compactor and test routine to run at the same pace. And we supply
the exact number of Advance() and wg.Wait() that compactor needs.

compactor/compactor_test.go

@@ -42,13 +42,17 @@ func TestPeriodic(t *testing.T) {
 	n := int(time.Hour / checkCompactionInterval)
 	// collect 3 hours of revisions
 	for i := 0; i < 3; i++ {
-		// advance one hour, one revision for each interval
-		for j := 0; j < n; j++ {
-			fc.Advance(checkCompactionInterval)
+		// advance one (hour - checkCompactionInterval), one revision for each interval
+		for j := 0; j < n-1; j++ {
 			_, err := rg.Wait(1)
 			if err != nil {
 				t.Fatal(err)
 			}
+			fc.Advance(checkCompactionInterval)
+		}
+		_, err := rg.Wait(1)
+		if err != nil {
+			t.Fatal(err)
 		}
 		// ready to acknowledge hour "i"
 		// block until compactor calls clock.After()
@@ -63,6 +67,12 @@ func TestPeriodic(t *testing.T) {
 			t.Errorf("compact request = %v, want %v", a[0].Params[0], &pb.CompactionRequest{Revision: int64(i*n) + 1})
 		}
 	}
+
+	// unblock the rev getter, so we can stop the compactor routine.
+	_, err := rg.Wait(1)
+	if err != nil {
+		t.Fatal(err)
+	}
 }
 
 func TestPeriodicPause(t *testing.T) {