You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by li...@apache.org on 2026/07/03 13:09:57 UTC
(skywalking-banyandb) 01/01: add Consistently check after Eventually checks
This is an automated email from the ASF dual-hosted git repository.
mrproliu pushed a commit to branch fix/add-consistently-checks
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 5015cbac33e6dd7dc0b76f58cf873db62e803075
Author: mrproliu <74...@qq.com>
AuthorDate: Fri Jul 3 21:09:42 2026 +0800
add Consistently check after Eventually checks
---
.../internal/migration/migration_stream_test.go | 6 +-
banyand/measure/metadata_test.go | 11 ++-
banyand/property/db/repair_gossip_test.go | 22 +++--
banyand/stream/metadata_test.go | 9 +-
banyand/stream/wqueue_test.go | 10 +-
banyand/trace/metadata_test.go | 7 +-
bydbctl/internal/cmd/measure_test.go | 5 +-
bydbctl/internal/cmd/property_test.go | 4 +-
bydbctl/internal/cmd/stream_test.go | 5 +-
bydbctl/internal/cmd/topn_test.go | 7 +-
bydbctl/internal/cmd/trace_test.go | 5 +-
pkg/test/eventually.go | 106 +++++++++++++++++++++
test/cases/lifecycle/lifecycle.go | 11 ++-
test/cases/lifecycle/orphan.go | 5 +-
test/cases/measure/measure.go | 3 +-
test/cases/property/property.go | 3 +-
test/cases/schema/clamp.go | 3 +-
test/cases/schema/shape_break.go | 7 +-
test/cases/schema/tag_family_move.go | 9 +-
test/cases/stream/stream.go | 3 +-
test/cases/trace/trace.go | 3 +-
test/integration/handoff/handoff_suite_test.go | 6 +-
.../replication/measure_normal_replication_test.go | 3 +-
test/integration/replication/replication_test.go | 3 +-
.../replication/stream_replication_test.go | 3 +-
.../replication/trace_replication_test.go | 3 +-
test/integration/standalone/other/measure_test.go | 3 +-
test/integration/standalone/other/tls_test.go | 7 +-
28 files changed, 205 insertions(+), 67 deletions(-)
diff --git a/banyand/internal/migration/migration_stream_test.go b/banyand/internal/migration/migration_stream_test.go
index 63a1fea17..662efe13c 100644
--- a/banyand/internal/migration/migration_stream_test.go
+++ b/banyand/internal/migration/migration_stream_test.go
@@ -255,7 +255,7 @@ func registerE2EIndexRule(t *testing.T, conn *grpc.ClientConn, group, subjectNam
// ready yet" (registry writes propagate to the serving modules asynchronously).
func awaitRows(t *testing.T, want int, fn func() (int, error)) {
t.Helper()
- gomega.Eventually(func() int {
+ test.EventuallyConsistently(func() int {
n, err := fn()
if err != nil {
return -1
@@ -270,7 +270,7 @@ func awaitRows(t *testing.T, want int, fn func() (int, error)) {
// registered instance, so plain one-shot assertions are CI-hostile.
func expectRows(t *testing.T, want int, explain string, fn func() (int, error)) {
t.Helper()
- gomega.Eventually(func() int {
+ test.EventuallyConsistently(func() int {
n, err := fn()
if err != nil {
return -1
@@ -289,7 +289,7 @@ var e2ePartDirPattern = regexp.MustCompile(`^[0-9a-f]{16}$`)
func awaitPartsOnDisk(t *testing.T, catalogDataRoot, group string, segStart time.Time) {
t.Helper()
segDir := filepath.Join(catalogDataRoot, group, "seg-"+segStart.Format("20060102"), "shard-0")
- gomega.Eventually(func() int {
+ test.EventuallyConsistently(func() int {
entries, err := os.ReadDir(segDir)
if err != nil {
return -1
diff --git a/banyand/measure/metadata_test.go b/banyand/measure/metadata_test.go
index ecb647581..501580bbe 100644
--- a/banyand/measure/metadata_test.go
+++ b/banyand/measure/metadata_test.go
@@ -36,6 +36,7 @@ import (
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
"github.com/apache/skywalking-banyandb/banyand/measure"
"github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
"github.com/apache/skywalking-banyandb/pkg/timestamp"
)
@@ -400,7 +401,7 @@ var _ = Describe("Schema Change", func() {
deleteExtraMeasureTag(svcs, measureName)
writeSchemaChangeMeasureData(svcs, measureName, now.Add(-1*time.Hour), 3, measureWriteDataOptions{entityIDPrefix: "entity_new_"})
- Eventually(func(innerGm Gomega) {
+ test.EventuallyConsistently(func(innerGm Gomega) {
dataPoints := querySchemaChangeMeasureData(svcs, measureName, now.Add(-3*time.Hour), now,
[]string{"id", "entity_id"}, []string{"total"})
innerGm.Expect(dataPoints).To(HaveLen(8))
@@ -512,7 +513,7 @@ var _ = Describe("Schema Change", func() {
deleteExtraMeasureTagFamily(svcs, measureName)
writeSchemaChangeMeasureData(svcs, measureName, now.Add(-1*time.Hour), 3, measureWriteDataOptions{entityIDPrefix: "entity_new_"})
- Eventually(func(innerGm Gomega) {
+ test.EventuallyConsistently(func(innerGm Gomega) {
dataPoints := querySchemaChangeMeasureData(svcs, measureName, now.Add(-3*time.Hour), now,
[]string{"id", "entity_id"}, []string{"total"})
innerGm.Expect(dataPoints).To(HaveLen(8))
@@ -540,7 +541,7 @@ var _ = Describe("Schema Change", func() {
deleteExtraMeasureTag(svcs, measureName)
writeSchemaChangeMeasureData(svcs, measureName, now.Add(-1*time.Hour), 3, measureWriteDataOptions{entityIDPrefix: "entity_new_"})
- Eventually(func(innerGm Gomega) {
+ test.EventuallyConsistently(func(innerGm Gomega) {
err := queryMeasureWithDeletedTagCondition(svcs, measureName, now)
innerGm.Expect(err).To(HaveOccurred())
innerGm.Expect(err.Error()).To(ContainSubstring("extra_tag"))
@@ -558,7 +559,7 @@ var _ = Describe("Schema Change", func() {
deleteExtraMeasureTag(svcs, measureName)
writeSchemaChangeMeasureData(svcs, measureName, now.Add(-1*time.Hour), 3, measureWriteDataOptions{entityIDPrefix: "entity_new_"})
- Eventually(func(innerGm Gomega) {
+ test.EventuallyConsistently(func(innerGm Gomega) {
err := queryMeasureWithDeletedTagProjection(svcs, measureName, now)
innerGm.Expect(err).To(HaveOccurred())
innerGm.Expect(err.Error()).To(ContainSubstring("extra_tag"))
@@ -576,7 +577,7 @@ var _ = Describe("Schema Change", func() {
deleteExtraField(svcs, measureName)
writeSchemaChangeMeasureData(svcs, measureName, now.Add(-1*time.Hour), 3, measureWriteDataOptions{entityIDPrefix: "entity_new_"})
- Eventually(func(innerGm Gomega) {
+ test.EventuallyConsistently(func(innerGm Gomega) {
err := queryMeasureWithDeletedFieldProjection(svcs, measureName, now)
innerGm.Expect(err).To(HaveOccurred())
innerGm.Expect(err.Error()).To(ContainSubstring("extra_field"))
diff --git a/banyand/property/db/repair_gossip_test.go b/banyand/property/db/repair_gossip_test.go
index 536f57920..3f169ea13 100644
--- a/banyand/property/db/repair_gossip_test.go
+++ b/banyand/property/db/repair_gossip_test.go
@@ -323,9 +323,17 @@ func startingEachTest(nodes *[]*nodeContext, ctrl *gomock.Controller, c *testCas
}
func startDataNodes(ctrl *gomock.Controller, nodes []node, groups []group) []*nodeContext {
+ // Allocate every node's gossip port in a single call so all ports are held
+ // simultaneously and are guaranteed distinct. Allocating one port per node
+ // separately releases each port before the async gossip server binds it,
+ // letting the OS hand the same just-released port to two nodes that then race
+ // to bind it — the loser hits "address already in use", the gossip server
+ // silently never comes up, and the readiness dial times out.
+ ports, err := test.AllocateFreePorts(len(nodes))
+ gomega.Expect(err).NotTo(gomega.HaveOccurred())
result := make([]*nodeContext, 0, len(nodes))
- for _, n := range nodes {
- result = append(result, startEachNode(ctrl, n, groups))
+ for i, n := range nodes {
+ result = append(result, startEachNode(ctrl, n, groups, ports[i]))
}
// registering the node in the gossip system
@@ -351,7 +359,7 @@ func startDataNodes(ctrl *gomock.Controller, nodes []node, groups []group) []*no
return result
}
-func startEachNode(ctrl *gomock.Controller, node node, groups []group) *nodeContext {
+func startEachNode(ctrl *gomock.Controller, node node, groups []group, port int) *nodeContext {
if node.treeSlotCount == 0 {
node.treeSlotCount = 32 // default value for tree slot count
}
@@ -382,12 +390,10 @@ func startEachNode(ctrl *gomock.Controller, node node, groups []group) *nodeCont
mockRepo.EXPECT().RegisterHandler("", schema.KindGroup, gomock.Any()).MaxTimes(1)
mockRepo.EXPECT().GroupRegistry().Return(mockGroup).AnyTimes()
- ports, err := test.AllocateFreePorts(1)
- gomega.Expect(err).NotTo(gomega.HaveOccurred())
messenger := gossip.NewMessengerWithoutMetadata("property-repair",
func(n *databasev1.Node) string { return n.PropertyRepairGossipGrpcAddress },
- observability.NewBypassRegistry(), ports[0])
- addr := fmt.Sprintf("127.0.0.1:%d", ports[0])
+ observability.NewBypassRegistry(), port)
+ addr := fmt.Sprintf("127.0.0.1:%d", port)
result.nodeID = addr
err = messenger.Validate()
gomega.Expect(err).NotTo(gomega.HaveOccurred())
@@ -531,7 +537,7 @@ func queryPropertyWithVerify(db *database, p property) {
}, groupField, entityID)
gomega.Expect(err).NotTo(gomega.HaveOccurred())
- gomega.Eventually(func() *property {
+ test.EventuallyConsistently(func() *property {
dataList, err := s.search(context.Background(), query, nil, 10)
if err != nil {
return nil
diff --git a/banyand/stream/metadata_test.go b/banyand/stream/metadata_test.go
index a0f0fc1b6..f813df7b2 100644
--- a/banyand/stream/metadata_test.go
+++ b/banyand/stream/metadata_test.go
@@ -35,6 +35,7 @@ import (
streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
"github.com/apache/skywalking-banyandb/banyand/stream"
"github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
"github.com/apache/skywalking-banyandb/pkg/timestamp"
)
@@ -395,7 +396,7 @@ var _ = Describe("Schema Change", func() {
deleteExtraTag(svcs, streamName, groupName)
writeSchemaChangeData(svcs, streamName, groupName, now.Add(-1*time.Hour), 3, writeDataOptions{})
- Eventually(func(innerGm Gomega) {
+ test.EventuallyConsistently(func(innerGm Gomega) {
elements := querySchemaChangeData(svcs, streamName, groupName, now.Add(-3*time.Hour), now,
[]string{"trace_id", "service_id", "duration"}, nil)
innerGm.Expect(elements).To(HaveLen(8))
@@ -507,7 +508,7 @@ var _ = Describe("Schema Change", func() {
deleteExtraTag(svcs, streamName, groupName)
writeSchemaChangeData(svcs, streamName, groupName, now.Add(-1*time.Hour), 3, writeDataOptions{})
- Eventually(func(innerGm Gomega) {
+ test.EventuallyConsistently(func(innerGm Gomega) {
err := queryWithDeletedTagCondition(svcs, streamName, groupName, now)
innerGm.Expect(err).To(HaveOccurred())
innerGm.Expect(err.Error()).To(ContainSubstring("extra_tag"))
@@ -525,7 +526,7 @@ var _ = Describe("Schema Change", func() {
deleteExtraTag(svcs, streamName, groupName)
writeSchemaChangeData(svcs, streamName, groupName, now.Add(-1*time.Hour), 3, writeDataOptions{})
- Eventually(func(innerGm Gomega) {
+ test.EventuallyConsistently(func(innerGm Gomega) {
err := queryWithDeletedTagProjection(svcs, streamName, groupName, now)
innerGm.Expect(err).To(HaveOccurred())
innerGm.Expect(err.Error()).To(ContainSubstring("extra_tag"))
@@ -545,7 +546,7 @@ var _ = Describe("Schema Change", func() {
deleteExtraTagFamily(svcs, streamName, groupName)
writeSchemaChangeData(svcs, streamName, groupName, now.Add(-1*time.Hour), 3, writeDataOptions{traceIDPrefix: "trace_new_", elementIDOffset: 5})
- Eventually(func(innerGm Gomega) {
+ test.EventuallyConsistently(func(innerGm Gomega) {
elements := querySchemaChangeData(svcs, streamName, groupName, now.Add(-3*time.Hour), now,
[]string{"trace_id", "service_id", "duration"}, nil)
innerGm.Expect(elements).To(HaveLen(8))
diff --git a/banyand/stream/wqueue_test.go b/banyand/stream/wqueue_test.go
index 8215e4c1c..09ca4a43b 100644
--- a/banyand/stream/wqueue_test.go
+++ b/banyand/stream/wqueue_test.go
@@ -79,10 +79,14 @@ func Test_newWriteQueue(t *testing.T) {
require.NotNil(t, tst)
defer tst.Close()
- // Wait for the creator to be set to snapshotCreatorSyncer
+ // Seed one batch of elements so the flush -> sync pipeline has a part to sync,
+ // then wait for the syncer to claim the snapshot. Re-adding elements on every
+ // poll (the previous approach) reset the snapshot creator back to
+ // snapshotCreatorMemPart each iteration, so a flush+sync that lagged the poll's
+ // 100ms window was never observed and this wait flaked under load. Adding once
+ // leaves the creator to settle on snapshotCreatorSyncer and stay there.
+ tst.mustAddElements(es)
assert.Eventually(t, func() bool {
- tst.mustAddElements(es)
- time.Sleep(100 * time.Millisecond)
s := tst.currentSnapshot()
if s == nil {
return false
diff --git a/banyand/trace/metadata_test.go b/banyand/trace/metadata_test.go
index 80faf468f..43d7ba42f 100644
--- a/banyand/trace/metadata_test.go
+++ b/banyand/trace/metadata_test.go
@@ -34,6 +34,7 @@ import (
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
tracev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/trace/v1"
"github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
"github.com/apache/skywalking-banyandb/pkg/timestamp"
)
@@ -268,7 +269,7 @@ var _ = Describe("Metadata", func() {
deleteTraceExtraTag(svcs, traceName, groupName)
writeSchemaChangeTraceData(svcs, traceName, groupName, now.Add(-1*time.Hour), 3, writeTraceDataOptions{})
- Eventually(func(innerGm Gomega) {
+ test.EventuallyConsistently(func(innerGm Gomega) {
spans := querySchemaChangeTraceData(svcs, traceName, groupName, now.Add(-3*time.Hour), now,
[]string{"trace_id", "service_id", "duration"})
innerGm.Expect(spans).To(HaveLen(8))
@@ -434,7 +435,7 @@ var _ = Describe("Metadata", func() {
deleteTraceExtraTag(svcs, traceName, groupName)
writeSchemaChangeTraceData(svcs, traceName, groupName, now.Add(-1*time.Hour), 3, writeTraceDataOptions{})
- Eventually(func(innerGm Gomega) {
+ test.EventuallyConsistently(func(innerGm Gomega) {
err := queryWithDeletedTagCondition(svcs, traceName, groupName, now)
innerGm.Expect(err).To(HaveOccurred())
innerGm.Expect(err.Error()).To(ContainSubstring("extra_tag"))
@@ -452,7 +453,7 @@ var _ = Describe("Metadata", func() {
deleteTraceExtraTag(svcs, traceName, groupName)
writeSchemaChangeTraceData(svcs, traceName, groupName, now.Add(-1*time.Hour), 3, writeTraceDataOptions{})
- Eventually(func(innerGm Gomega) {
+ test.EventuallyConsistently(func(innerGm Gomega) {
err := queryWithDeletedTagProjection(svcs, traceName, groupName, now)
innerGm.Expect(err).To(HaveOccurred())
innerGm.Expect(err.Error()).To(ContainSubstring("extra_tag"))
diff --git a/bydbctl/internal/cmd/measure_test.go b/bydbctl/internal/cmd/measure_test.go
index 288a3f890..f93610cd4 100644
--- a/bydbctl/internal/cmd/measure_test.go
+++ b/bydbctl/internal/cmd/measure_test.go
@@ -32,6 +32,7 @@ import (
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
"github.com/apache/skywalking-banyandb/bydbctl/internal/cmd"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
"github.com/apache/skywalking-banyandb/pkg/test/helpers"
"github.com/apache/skywalking-banyandb/pkg/test/setup"
@@ -262,7 +263,7 @@ tagProjection:
return buf.String()
}
Eventually(issue, flags.EventuallyTimeout).ShouldNot(ContainSubstring("code:"))
- Eventually(func() int {
+ test.EventuallyConsistently(func() int {
out := issue()
GinkgoWriter.Println(out)
resp := new(measurev1.QueryResponse)
@@ -303,7 +304,7 @@ tagProjection:
return buf.String()
}
Eventually(issue, flags.EventuallyTimeout).ShouldNot(ContainSubstring("code:"))
- Eventually(func() int {
+ test.EventuallyConsistently(func() int {
out := issue()
resp := new(measurev1.QueryResponse)
helpers.UnmarshalYAML([]byte(out), resp)
diff --git a/bydbctl/internal/cmd/property_test.go b/bydbctl/internal/cmd/property_test.go
index 7377cc094..4ce3d47ab 100644
--- a/bydbctl/internal/cmd/property_test.go
+++ b/bydbctl/internal/cmd/property_test.go
@@ -717,7 +717,7 @@ projection:
return buf.String()
}
Eventually(issue, flags.EventuallyTimeout).ShouldNot(ContainSubstring("code:"))
- Eventually(func() int {
+ test.EventuallyConsistently(func() int {
out := issue()
GinkgoWriter.Println(out)
resp := new(streamv1.QueryResponse)
@@ -1036,7 +1036,7 @@ func queryData(rootCmd *cobra.Command, addr, group, id string, dataCount int, ve
Expect(err).NotTo(HaveOccurred())
return buf.String()
}
- Eventually(func() error {
+ test.EventuallyConsistently(func() error {
out := issue()
resp := new(propertyv1.QueryResponse)
helpers.UnmarshalYAML([]byte(out), resp)
diff --git a/bydbctl/internal/cmd/stream_test.go b/bydbctl/internal/cmd/stream_test.go
index 5bb082945..5dcc0be6c 100644
--- a/bydbctl/internal/cmd/stream_test.go
+++ b/bydbctl/internal/cmd/stream_test.go
@@ -32,6 +32,7 @@ import (
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
"github.com/apache/skywalking-banyandb/bydbctl/internal/cmd"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
"github.com/apache/skywalking-banyandb/pkg/test/helpers"
"github.com/apache/skywalking-banyandb/pkg/test/setup"
@@ -263,7 +264,7 @@ projection:
return buf.String()
}
Eventually(issue, flags.EventuallyTimeout).ShouldNot(ContainSubstring("code:"))
- Eventually(func() int {
+ test.EventuallyConsistently(func() int {
out := issue()
resp := new(streamv1.QueryResponse)
helpers.UnmarshalYAML([]byte(out), resp)
@@ -303,7 +304,7 @@ projection:
return buf.String()
}
Eventually(issue, flags.EventuallyTimeout).ShouldNot(ContainSubstring("code:"))
- Eventually(func() int {
+ test.EventuallyConsistently(func() int {
out := issue()
resp := new(streamv1.QueryResponse)
helpers.UnmarshalYAML([]byte(out), resp)
diff --git a/bydbctl/internal/cmd/topn_test.go b/bydbctl/internal/cmd/topn_test.go
index 0413541d1..5e7509b4d 100644
--- a/bydbctl/internal/cmd/topn_test.go
+++ b/bydbctl/internal/cmd/topn_test.go
@@ -32,6 +32,7 @@ import (
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
"github.com/apache/skywalking-banyandb/bydbctl/internal/cmd"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
"github.com/apache/skywalking-banyandb/pkg/test/helpers"
"github.com/apache/skywalking-banyandb/pkg/test/setup"
@@ -312,7 +313,7 @@ tagProjection:
return buf.String()
}
Eventually(issue, flags.EventuallyTimeout).ShouldNot(ContainSubstring("code:"))
- Eventually(func() int {
+ test.EventuallyConsistently(func() int {
out := issue()
GinkgoWriter.Println(out)
resp := new(measurev1.QueryResponse)
@@ -341,7 +342,7 @@ fieldValueSort: 1`, startStr, endStr)))
return buf.String()
}
Eventually(issue1, flags.EventuallyTimeout).ShouldNot(ContainSubstring("code:"))
- Eventually(func() int {
+ test.EventuallyConsistently(func() int {
out := issue1()
GinkgoWriter.Println(out)
resp := new(measurev1.TopNResponse)
@@ -384,7 +385,7 @@ fieldValueSort: 1`))
return buf.String()
}
Eventually(issue1, flags.EventuallyTimeout).ShouldNot(ContainSubstring("code:"))
- Eventually(func() int {
+ test.EventuallyConsistently(func() int {
out := issue1()
GinkgoWriter.Println(out)
resp := new(measurev1.TopNResponse)
diff --git a/bydbctl/internal/cmd/trace_test.go b/bydbctl/internal/cmd/trace_test.go
index 6f503faa7..b82f42ae6 100644
--- a/bydbctl/internal/cmd/trace_test.go
+++ b/bydbctl/internal/cmd/trace_test.go
@@ -32,6 +32,7 @@ import (
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
tracev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/trace/v1"
"github.com/apache/skywalking-banyandb/bydbctl/internal/cmd"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
"github.com/apache/skywalking-banyandb/pkg/test/helpers"
"github.com/apache/skywalking-banyandb/pkg/test/setup"
@@ -277,7 +278,7 @@ orderBy:
return buf.String()
}
Eventually(issue, flags.EventuallyTimeout).ShouldNot(ContainSubstring("code:"))
- Eventually(func() int {
+ test.EventuallyConsistently(func() int {
out := issue()
resp := new(tracev1.QueryResponse)
helpers.UnmarshalYAML([]byte(out), resp)
@@ -316,7 +317,7 @@ orderBy:
return buf.String()
}
Eventually(issue, flags.EventuallyTimeout).ShouldNot(ContainSubstring("code:"))
- Eventually(func() int {
+ test.EventuallyConsistently(func() int {
out := issue()
resp := new(tracev1.QueryResponse)
helpers.UnmarshalYAML([]byte(out), resp)
diff --git a/pkg/test/eventually.go b/pkg/test/eventually.go
new file mode 100644
index 000000000..8c2b9133d
--- /dev/null
+++ b/pkg/test/eventually.go
@@ -0,0 +1,106 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package test
+
+import (
+ "time"
+
+ "github.com/onsi/gomega"
+ gomegatypes "github.com/onsi/gomega/types"
+
+ "github.com/apache/skywalking-banyandb/pkg/test/flags"
+)
+
+// ECAssertion pairs a gomega Eventually with a following Consistently. It mirrors
+// gomega.Eventually's layout: intervals may be supplied up front or via the
+// fluent WithTimeout/WithPolling methods, and the matcher is applied via
+// Should/ShouldNot.
+type ECAssertion struct {
+ actual interface{}
+ eventuallyTimeout interface{}
+ polling interface{}
+}
+
+// EventuallyConsistently returns an assertion that first waits for actual to
+// satisfy the matcher (Eventually) and then requires it to keep satisfying the
+// matcher for a stability window (Consistently). Pairing convergence with a
+// stability window catches transient states that briefly match and then regress,
+// which a lone Eventually would accept.
+//
+// intervals follow gomega.Eventually semantics — (timeout) or (timeout, polling).
+// They can also be set fluently via WithTimeout/WithPolling. When the timeout is
+// omitted it defaults to flags.EventuallyTimeout; the Consistently window always
+// uses flags.ConsistentlyTimeout and reuses the polling interval, if any.
+func EventuallyConsistently(actual interface{}, intervals ...interface{}) ECAssertion {
+ a := ECAssertion{actual: actual}
+ if len(intervals) > 0 {
+ a.eventuallyTimeout = intervals[0]
+ }
+ if len(intervals) > 1 {
+ a.polling = intervals[1]
+ }
+ return a
+}
+
+// WithTimeout overrides the Eventually convergence timeout, mirroring
+// gomega.AsyncAssertion.WithTimeout.
+func (a ECAssertion) WithTimeout(timeout time.Duration) ECAssertion {
+ a.eventuallyTimeout = timeout
+ return a
+}
+
+// WithPolling overrides the polling interval used by both the Eventually and the
+// Consistently phase, mirroring gomega.AsyncAssertion.WithPolling.
+func (a ECAssertion) WithPolling(polling time.Duration) ECAssertion {
+ a.polling = polling
+ return a
+}
+
+// Should asserts the matcher is reached and then held. optionalDescription is
+// forwarded to both assertions, matching gomega's Should.
+func (a ECAssertion) Should(matcher gomegatypes.GomegaMatcher, optionalDescription ...interface{}) {
+ gomega.EventuallyWithOffset(1, a.actual, a.eventuallyIntervals()...).Should(matcher, optionalDescription...)
+ gomega.ConsistentlyWithOffset(1, a.actual, a.consistentlyIntervals()...).Should(matcher, optionalDescription...)
+}
+
+// ShouldNot asserts the matcher is avoided and then kept avoided. optionalDescription
+// is forwarded to both assertions, matching gomega's ShouldNot.
+func (a ECAssertion) ShouldNot(matcher gomegatypes.GomegaMatcher, optionalDescription ...interface{}) {
+ gomega.EventuallyWithOffset(1, a.actual, a.eventuallyIntervals()...).ShouldNot(matcher, optionalDescription...)
+ gomega.ConsistentlyWithOffset(1, a.actual, a.consistentlyIntervals()...).ShouldNot(matcher, optionalDescription...)
+}
+
+func (a ECAssertion) eventuallyIntervals() []interface{} {
+ timeout := a.eventuallyTimeout
+ if timeout == nil {
+ timeout = flags.EventuallyTimeout
+ }
+ out := []interface{}{timeout}
+ if a.polling != nil {
+ out = append(out, a.polling)
+ }
+ return out
+}
+
+func (a ECAssertion) consistentlyIntervals() []interface{} {
+ out := []interface{}{flags.ConsistentlyTimeout}
+ if a.polling != nil {
+ out = append(out, a.polling)
+ }
+ return out
+}
diff --git a/test/cases/lifecycle/lifecycle.go b/test/cases/lifecycle/lifecycle.go
index 11147bd2e..a127a96b5 100644
--- a/test/cases/lifecycle/lifecycle.go
+++ b/test/cases/lifecycle/lifecycle.go
@@ -47,6 +47,7 @@ import (
"github.com/apache/skywalking-banyandb/banyand/backup/lifecycle"
"github.com/apache/skywalking-banyandb/banyand/observability"
"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
"github.com/apache/skywalking-banyandb/pkg/test/helpers"
measureTestData "github.com/apache/skywalking-banyandb/test/cases/measure/data"
@@ -198,7 +199,7 @@ func verifyLifecycleStages(sc helpers.SharedContext, verifyFn func(gomega.Gomega
})
// Verify hot+warm stages exist after migration
- gomega.Eventually(func(innerGm gomega.Gomega) {
+ test.EventuallyConsistently(func(innerGm gomega.Gomega) {
verifyFn(innerGm, sc, helpers.Args{
Input: args.Input,
Duration: args.Duration,
@@ -209,7 +210,7 @@ func verifyLifecycleStages(sc helpers.SharedContext, verifyFn func(gomega.Gomega
}, flags.EventuallyTimeout).Should(gomega.Succeed())
// Verify warm stage only after retention
- gomega.Eventually(func(innerGm gomega.Gomega) {
+ test.EventuallyConsistently(func(innerGm gomega.Gomega) {
verifyFn(innerGm, sc, helpers.Args{
Input: args.Input,
Duration: args.Duration,
@@ -965,7 +966,7 @@ var _ = ginkgo.Describe("Measure cross-segment migration", ginkgo.Ordered, func(
Stages: []string{"warm"},
}
var resp *measurev1.QueryResponse
- gomega.Eventually(func() error {
+ test.EventuallyConsistently(func() error {
var qErr error
resp, qErr = queryClient.Query(ctx, req)
if qErr != nil {
@@ -1254,7 +1255,7 @@ var _ = ginkgo.Describe("Stream cross-segment migration", ginkgo.Ordered, func()
Stages: []string{"warm"},
}
var resp *streamv1.QueryResponse
- gomega.Eventually(func() error {
+ test.EventuallyConsistently(func() error {
var qErr error
resp, qErr = queryClient.Query(ctx, req)
if qErr != nil {
@@ -1426,7 +1427,7 @@ var _ = ginkgo.Describe("Trace cross-segment migration", ginkgo.Ordered, func()
Stages: []string{"warm"},
}
var resp *tracev1.QueryResponse
- gomega.Eventually(func() error {
+ test.EventuallyConsistently(func() error {
var qErr error
resp, qErr = queryClient.Query(ctx, req)
if qErr != nil {
diff --git a/test/cases/lifecycle/orphan.go b/test/cases/lifecycle/orphan.go
index fa59d7387..bb2fb0a90 100644
--- a/test/cases/lifecycle/orphan.go
+++ b/test/cases/lifecycle/orphan.go
@@ -44,6 +44,7 @@ import (
streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
"github.com/apache/skywalking-banyandb/banyand/backup/lifecycle"
"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
)
@@ -316,7 +317,7 @@ var _ = ginkgo.Describe("Lifecycle orphan-schema archive", ginkgo.Ordered, func(
ginkgo.By("verifying the kept measure migrated and is queryable on the warm stage")
queryClient := measurev1.NewMeasureServiceClient(conn)
- gomega.Eventually(func() error {
+ test.EventuallyConsistently(func() error {
resp, qErr := queryClient.Query(ctx, &measurev1.QueryRequest{
Groups: []string{group}, Name: keep,
TimeRange: &modelv1.TimeRange{Begin: timestamppb.New(leftTS.Add(-time.Hour)), End: timestamppb.New(rightTS.Add(time.Hour))},
@@ -458,7 +459,7 @@ var _ = ginkgo.Describe("Lifecycle orphan-schema archive", ginkgo.Ordered, func(
ginkgo.By("verifying the kept stream migrated and is queryable on the warm stage")
queryClient := streamv1.NewStreamServiceClient(conn)
- gomega.Eventually(func() error {
+ test.EventuallyConsistently(func() error {
resp, qErr := queryClient.Query(ctx, &streamv1.QueryRequest{
Groups: []string{group}, Name: keep,
TimeRange: &modelv1.TimeRange{Begin: timestamppb.New(leftTS.Add(-time.Hour)), End: timestamppb.New(rightTS.Add(time.Hour))},
diff --git a/test/cases/measure/measure.go b/test/cases/measure/measure.go
index 20ec2da31..4594dfeb6 100644
--- a/test/cases/measure/measure.go
+++ b/test/cases/measure/measure.go
@@ -24,6 +24,7 @@ import (
g "github.com/onsi/ginkgo/v2"
gm "github.com/onsi/gomega"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
"github.com/apache/skywalking-banyandb/pkg/test/helpers"
measureTestData "github.com/apache/skywalking-banyandb/test/cases/measure/data"
@@ -33,7 +34,7 @@ var (
// SharedContext is the parallel execution context.
SharedContext helpers.SharedContext
verify = func(args helpers.Args) {
- gm.Eventually(func(innerGm gm.Gomega) {
+ test.EventuallyConsistently(func(innerGm gm.Gomega) {
measureTestData.VerifyFn(innerGm, SharedContext, args)
}, flags.EventuallyTimeout).Should(gm.Succeed())
}
diff --git a/test/cases/property/property.go b/test/cases/property/property.go
index 9b8d2efb5..36cb20c87 100644
--- a/test/cases/property/property.go
+++ b/test/cases/property/property.go
@@ -22,6 +22,7 @@ import (
g "github.com/onsi/ginkgo/v2"
gm "github.com/onsi/gomega"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
"github.com/apache/skywalking-banyandb/pkg/test/helpers"
propertyTestData "github.com/apache/skywalking-banyandb/test/cases/property/data"
@@ -31,7 +32,7 @@ var (
// SharedContext is the parallel execution context.
SharedContext helpers.SharedContext
verify = func(args helpers.Args) {
- gm.Eventually(func(innerGm gm.Gomega) {
+ test.EventuallyConsistently(func(innerGm gm.Gomega) {
propertyTestData.VerifyFn(innerGm, SharedContext, args)
}, flags.EventuallyTimeout).Should(gm.Succeed())
}
diff --git a/test/cases/schema/clamp.go b/test/cases/schema/clamp.go
index a23144945..2e113bdd7 100644
--- a/test/cases/schema/clamp.go
+++ b/test/cases/schema/clamp.go
@@ -31,6 +31,7 @@ import (
measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/timestamp"
)
@@ -248,7 +249,7 @@ var _ = g.Describe("Schema time-range clamp", func() {
// data-node ack returns when mustAddMemPart's applied channel
// fires, but the query fan-out can still race the new memPart
// becoming visible. Mirrors the deletion.go retry pattern.
- gm.Eventually(func() int {
+ test.EventuallyConsistently(func() int {
baselineResp, baselineErr := queryMeasureRange(ctx, clients.MeasureWriteClient, group1, measureName,
tData1.Add(-time.Hour), time.Now().Add(time.Hour), 0)
if baselineErr != nil {
diff --git a/test/cases/schema/shape_break.go b/test/cases/schema/shape_break.go
index b939adf27..59ca581bc 100644
--- a/test/cases/schema/shape_break.go
+++ b/test/cases/schema/shape_break.go
@@ -31,6 +31,7 @@ import (
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/timestamp"
)
@@ -209,7 +210,7 @@ var _ = g.Describe("Schema shape-break rejection", func() {
// schema-consistency suite already uses this Eventually pattern
// in deletion.go for the same reason; mirror it here.
var queryResp1 *measurev1.QueryResponse
- gm.Eventually(func() int {
+ test.EventuallyConsistently(func() int {
var queryErr1 error
queryResp1, queryErr1 = queryMeasureRange(ctx, clients.MeasureWriteClient, groupName, measureName,
createdAt1.AsTime(), time.Now().Add(time.Hour), r1)
@@ -455,7 +456,7 @@ var _ = g.Describe("Schema shape-break rejection", func() {
// Eventually retry — see the baseline above for the
// distributed write→query visibility race.
- gm.Eventually(func() int {
+ test.EventuallyConsistently(func() int {
queryResp1, queryErr1 := queryMeasureRange(ctx, clients.MeasureWriteClient, groupName, measureName,
createdAt1.AsTime(), time.Now().Add(time.Hour), r1)
if queryErr1 != nil {
@@ -516,7 +517,7 @@ var _ = g.Describe("Schema shape-break rejection", func() {
// Post-write query [CreatedAt2, now+1h] must return the newly-written point.
// Same write→query visibility race as the baseline.
- gm.Eventually(func() int {
+ test.EventuallyConsistently(func() int {
queryResp3, queryErr3 := queryMeasureRange(ctx, clients.MeasureWriteClient, groupName, measureName,
createdAt2.AsTime(), time.Now().Add(time.Hour), r2)
if queryErr3 != nil {
diff --git a/test/cases/schema/tag_family_move.go b/test/cases/schema/tag_family_move.go
index fae69ebc3..0f12d95e6 100644
--- a/test/cases/schema/tag_family_move.go
+++ b/test/cases/schema/tag_family_move.go
@@ -33,6 +33,7 @@ import (
measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
modelv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/model/v1"
streamv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/stream/v1"
+ "github.com/apache/skywalking-banyandb/pkg/test"
)
const (
@@ -565,7 +566,7 @@ func queryTagFamilyMoveStream(
}
}
var resp *streamv1.QueryResponse
- gm.Eventually(func() int {
+ test.EventuallyConsistently(func() int {
queryResp, queryErr := client.Query(ctx, &streamv1.QueryRequest{
Groups: []string{groupName},
Name: "tfm_stream",
@@ -610,7 +611,7 @@ func queryStorageToSearchableMoveStream(
}
}
var resp *streamv1.QueryResponse
- gm.Eventually(func() int {
+ test.EventuallyConsistently(func() int {
queryResp, queryErr := client.Query(ctx, &streamv1.QueryRequest{
Groups: []string{groupName},
Name: "tfm_storage_stream",
@@ -684,7 +685,7 @@ func queryTagFamilyMoveMeasure(
) []*measurev1.DataPoint {
projection := tagFamilyMoveMeasureProjection(movedProjection)
var resp *measurev1.QueryResponse
- gm.Eventually(func() int {
+ test.EventuallyConsistently(func() int {
queryResp, queryErr := client.Query(ctx, &measurev1.QueryRequest{
Groups: []string{groupName},
Name: "tfm_measure",
@@ -719,7 +720,7 @@ func queryStorageToSearchableMoveMeasure(
) []*measurev1.DataPoint {
projection := tagFamilyMoveMeasureStorageToSearchableProjection(movedProjection)
var resp *measurev1.QueryResponse
- gm.Eventually(func() int {
+ test.EventuallyConsistently(func() int {
queryResp, queryErr := client.Query(ctx, &measurev1.QueryRequest{
Groups: []string{groupName},
Name: "tfm_storage_measure",
diff --git a/test/cases/stream/stream.go b/test/cases/stream/stream.go
index 79ea80330..a5520e420 100644
--- a/test/cases/stream/stream.go
+++ b/test/cases/stream/stream.go
@@ -26,6 +26,7 @@ import (
gm "github.com/onsi/gomega"
"google.golang.org/protobuf/types/known/timestamppb"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
"github.com/apache/skywalking-banyandb/pkg/test/helpers"
stream_test_data "github.com/apache/skywalking-banyandb/test/cases/stream/data"
@@ -105,7 +106,7 @@ var streamEntries = []any{
// RegisterTable registers the stream test table with the given description.
func RegisterTable(description string) bool {
return g.DescribeTable(description, append([]any{func(args helpers.Args) {
- gm.Eventually(func(innerGm gm.Gomega) {
+ test.EventuallyConsistently(func(innerGm gm.Gomega) {
verify(innerGm, args)
}, flags.EventuallyTimeout).Should(gm.Succeed())
}}, streamEntries...)...)
diff --git a/test/cases/trace/trace.go b/test/cases/trace/trace.go
index fdc14527f..e3afe26d5 100644
--- a/test/cases/trace/trace.go
+++ b/test/cases/trace/trace.go
@@ -24,6 +24,7 @@ import (
g "github.com/onsi/ginkgo/v2"
gm "github.com/onsi/gomega"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
"github.com/apache/skywalking-banyandb/pkg/test/helpers"
trace_test_data "github.com/apache/skywalking-banyandb/test/cases/trace/data"
@@ -118,7 +119,7 @@ var traceEntries = []any{
// RegisterTable registers the trace test table with the given description.
func RegisterTable(description string) bool {
return g.DescribeTable(description, append([]any{func(args helpers.Args) {
- gm.Eventually(func(innerGm gm.Gomega) {
+ test.EventuallyConsistently(func(innerGm gm.Gomega) {
verify(innerGm, args)
}, flags.EventuallyTimeout).Should(gm.Succeed())
}}, traceEntries...)...)
diff --git a/test/integration/handoff/handoff_suite_test.go b/test/integration/handoff/handoff_suite_test.go
index e1af1ca70..0a097cb3d 100644
--- a/test/integration/handoff/handoff_suite_test.go
+++ b/test/integration/handoff/handoff_suite_test.go
@@ -366,12 +366,12 @@ var _ = Describe("trace handoff", func() {
}, flags.EventuallyTimeout).Should(Equal(0))
By("verifying the replayed trace can be queried")
- Eventually(func() error {
+ test.EventuallyConsistently(func() error {
return queryTrace(connection, traceID, writeTime)
}, flags.EventuallyTimeout).Should(Succeed())
By("verifying sidx data was replayed correctly by querying via indexed tag")
- Eventually(func() error {
+ test.EventuallyConsistently(func() error {
return queryTraceByService(connection, "handoff_service", writeTime)
}, flags.EventuallyTimeout).Should(Succeed())
@@ -387,7 +387,7 @@ var _ = Describe("trace handoff", func() {
dnHandles[otherIndex].stop()
defer dnHandles[otherIndex].start(discoveryFilePath)
- Eventually(func() error {
+ test.EventuallyConsistently(func() error {
return queryTrace(connection, traceID, writeTime)
}, flags.EventuallyTimeout).Should(Succeed())
diff --git a/test/integration/replication/measure_normal_replication_test.go b/test/integration/replication/measure_normal_replication_test.go
index d2f2857ff..4852c0db2 100644
--- a/test/integration/replication/measure_normal_replication_test.go
+++ b/test/integration/replication/measure_normal_replication_test.go
@@ -29,6 +29,7 @@ import (
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
"github.com/apache/skywalking-banyandb/pkg/test/helpers"
"github.com/apache/skywalking-banyandb/pkg/test/setup"
@@ -160,7 +161,7 @@ func verifyDataContentWithArgs(conn *grpc.ClientConn, baseTime time.Time, args h
Connection: conn,
BaseTime: baseTime,
}
- gm.Eventually(func(innerGm gm.Gomega) {
+ test.EventuallyConsistently(func(innerGm gm.Gomega) {
casesmeasuredata.VerifyFn(innerGm, sharedContext, args)
}, flags.EventuallyTimeout).Should(gm.Succeed(),
"Should be able to query and verify data content")
diff --git a/test/integration/replication/replication_test.go b/test/integration/replication/replication_test.go
index 020a9c45f..1636376d7 100644
--- a/test/integration/replication/replication_test.go
+++ b/test/integration/replication/replication_test.go
@@ -29,6 +29,7 @@ import (
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
"github.com/apache/skywalking-banyandb/pkg/test/helpers"
casesmeasuredata "github.com/apache/skywalking-banyandb/test/cases/measure/data"
@@ -116,7 +117,7 @@ func verifyDataContentAfterNodeFailure(conn *grpc.ClientConn, baseTime time.Time
// 1. Read entity_replicated.ql and entity_replicated.yaml
// 2. Execute the query
// 3. Verify results match expected data
- gm.Eventually(func(innerGm gm.Gomega) {
+ test.EventuallyConsistently(func(innerGm gm.Gomega) {
casesmeasuredata.VerifyFn(innerGm, sharedContext, args)
}, flags.EventuallyTimeout).Should(gm.Succeed(),
"Should be able to query and verify data content after node failure")
diff --git a/test/integration/replication/stream_replication_test.go b/test/integration/replication/stream_replication_test.go
index b4ebf6460..c89b67588 100644
--- a/test/integration/replication/stream_replication_test.go
+++ b/test/integration/replication/stream_replication_test.go
@@ -29,6 +29,7 @@ import (
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
"github.com/apache/skywalking-banyandb/pkg/test/helpers"
"github.com/apache/skywalking-banyandb/pkg/test/setup"
@@ -160,7 +161,7 @@ func verifyStreamDataContentWithArgs(conn *grpc.ClientConn, baseTime time.Time,
Connection: conn,
BaseTime: baseTime,
}
- gm.Eventually(func(innerGm gm.Gomega) {
+ test.EventuallyConsistently(func(innerGm gm.Gomega) {
casesstreamdata.VerifyFn(innerGm, sharedContext, args)
}, flags.EventuallyTimeout).Should(gm.Succeed(),
"Should be able to query and verify stream data content")
diff --git a/test/integration/replication/trace_replication_test.go b/test/integration/replication/trace_replication_test.go
index 199f88b0b..1f66b4a9a 100644
--- a/test/integration/replication/trace_replication_test.go
+++ b/test/integration/replication/trace_replication_test.go
@@ -29,6 +29,7 @@ import (
commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
"github.com/apache/skywalking-banyandb/pkg/test/helpers"
"github.com/apache/skywalking-banyandb/pkg/test/setup"
@@ -161,7 +162,7 @@ func verifyTraceDataContentWithArgs(conn *grpc.ClientConn, baseTime time.Time, a
Connection: conn,
BaseTime: baseTime,
}
- gm.Eventually(func(innerGm gm.Gomega) {
+ test.EventuallyConsistently(func(innerGm gm.Gomega) {
casestracedata.VerifyFn(innerGm, sharedContext, args)
}, flags.EventuallyTimeout).Should(gm.Succeed(),
"Should be able to query and verify trace data content")
diff --git a/test/integration/standalone/other/measure_test.go b/test/integration/standalone/other/measure_test.go
index bdd7b331f..b08013502 100644
--- a/test/integration/standalone/other/measure_test.go
+++ b/test/integration/standalone/other/measure_test.go
@@ -28,6 +28,7 @@ import (
"github.com/apache/skywalking-banyandb/pkg/grpchelper"
"github.com/apache/skywalking-banyandb/pkg/pool"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
"github.com/apache/skywalking-banyandb/pkg/test/gmatcher"
"github.com/apache/skywalking-banyandb/pkg/test/helpers"
@@ -63,7 +64,7 @@ var _ = g.Describe("Query service_cpm_minute", func() {
})
g.It("queries service_cpm_minute by id after updating", func() {
casesMeasureData.Write(conn, "service_cpm_minute", "sw_metric", "service_cpm_minute_data1.json", baseTime, interval)
- gm.Eventually(func(innerGm gm.Gomega) {
+ test.EventuallyConsistently(func(innerGm gm.Gomega) {
casesMeasureData.VerifyFn(innerGm, helpers.SharedContext{
Connection: conn,
BaseTime: baseTime,
diff --git a/test/integration/standalone/other/tls_test.go b/test/integration/standalone/other/tls_test.go
index 7269ea12b..76d09f50e 100644
--- a/test/integration/standalone/other/tls_test.go
+++ b/test/integration/standalone/other/tls_test.go
@@ -35,6 +35,7 @@ import (
"google.golang.org/grpc/credentials"
"github.com/apache/skywalking-banyandb/pkg/grpchelper"
+ "github.com/apache/skywalking-banyandb/pkg/test"
"github.com/apache/skywalking-banyandb/pkg/test/flags"
"github.com/apache/skywalking-banyandb/pkg/test/helpers"
"github.com/apache/skywalking-banyandb/pkg/test/setup"
@@ -75,7 +76,7 @@ var _ = g.Describe("Query service_cpm_minute", func() {
gm.Eventually(gleak.Goroutines, flags.EventuallyTimeout).ShouldNot(gleak.HaveLeaked(goods))
})
g.It("queries a tls server", func() {
- gm.Eventually(func(innerGm gm.Gomega) {
+ test.EventuallyConsistently(func(innerGm gm.Gomega) {
casesMeasureData.VerifyFn(innerGm, helpers.SharedContext{
Connection: conn,
BaseTime: baseTime,
@@ -157,7 +158,7 @@ var _ = g.Describe("Query service_cpm_minute", func() {
casesMeasureData.Write(grpcConn, "service_cpm_minute", "sw_metric", "service_cpm_minute_data.json", testBaseTime, interval)
// Verify using the initial connection before updating certificates
- gm.Eventually(func(innerGm gm.Gomega) {
+ test.EventuallyConsistently(func(innerGm gm.Gomega) {
casesMeasureData.VerifyFn(innerGm, helpers.SharedContext{
Connection: grpcConn,
BaseTime: testBaseTime,
@@ -221,7 +222,7 @@ var _ = g.Describe("Query service_cpm_minute", func() {
defer newGrpcConn.Close()
// Verify data access works through a new connection
- gm.Eventually(func(innerGm gm.Gomega) {
+ test.EventuallyConsistently(func(innerGm gm.Gomega) {
casesMeasureData.VerifyFn(innerGm, helpers.SharedContext{
Connection: newGrpcConn,
BaseTime: testBaseTime,