This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new a4fac2f4b Add Consistently check after Eventually checks (#1204)
a4fac2f4b is described below
commit a4fac2f4bd82e5fcf124ce9a7ef61173802b051b
Author: mrproliu <[email protected]>
AuthorDate: Mon Jul 6 11:23:16 2026 +0800
Add Consistently check after Eventually checks (#1204)
---
.../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 | 31 +++--
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 | 151 +++++++++++++++++++++
pkg/test/eventually_test.go | 62 +++++++++
pkg/test/flags/flags.go | 31 +++++
test/cases/lifecycle/lifecycle.go | 25 ++--
test/cases/lifecycle/orphan.go | 5 +-
test/cases/measure/measure.go | 3 +-
test/cases/property/data/data.go | 5 +
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/cmd/generate/seed.go | 4 +-
.../cases/trace/data/testdata/sw_mixed_traces.json | 4 +-
...duration_asc_limit2_offset1_proj_explicit_1.yml | 38 +++---
...t_order_duration_asc_limit5_proj_explicit_6.yml | 78 +++++------
...en_feat_order_duration_desc_proj_explicit_3.yml | 4 +-
...ter_state_eq_limit5_offset1_proj_explicit_8.yml | 4 +-
test/cases/trace/data/want/gen_leaf_eq_state.yml | 4 +-
.../cases/trace/data/want/gen_leaf_ge_duration.yml | 4 +-
test/cases/trace/data/want/gen_leaf_ge_state.yml | 4 +-
.../cases/trace/data/want/gen_leaf_gt_duration.yml | 4 +-
test/cases/trace/data/want/gen_leaf_gt_state.yml | 4 +-
.../trace/data/want/gen_leaf_in_service_id.yml | 2 +-
.../cases/trace/data/want/gen_leaf_in_trace_id.yml | 2 +-
.../cases/trace/data/want/gen_leaf_le_duration.yml | 4 +-
test/cases/trace/data/want/gen_leaf_le_state.yml | 4 +-
.../cases/trace/data/want/gen_leaf_lt_duration.yml | 4 +-
test/cases/trace/data/want/gen_leaf_lt_state.yml | 4 +-
.../trace/data/want/gen_leaf_ne_service_id.yml | 4 +-
test/cases/trace/data/want/gen_leaf_ne_state.yml | 4 +-
.../cases/trace/data/want/gen_leaf_ne_trace_id.yml | 4 +-
.../trace/data/want/gen_leaf_not_in_service_id.yml | 4 +-
.../trace/data/want/gen_leaf_not_in_trace_id.yml | 4 +-
.../trace/data/want/gen_tree_depth5_deep_and.yml | 4 +-
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 +-
54 files changed, 476 insertions(+), 176 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..15a403996 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)
@@ -978,7 +978,12 @@ resource_opts:
}
return buf.String()
}
- Eventually(createGroup,
flags.EventuallyTimeout).Should(ContainSubstring("group ui-template is
created"))
+ Eventually(createGroup, flags.EventuallyTimeout).Should(
+ Or(ContainSubstring("group ui-template is created"),
ContainSubstring("already exists")))
+ // A reused data dir may have re-registered the group on startup; update
+ // unconditionally so the desired shard/replica opts always win.
+ rootCmd.SetArgs([]string{"group", "update", "-a", addr, "-f", "-"})
+ Eventually(createGroup,
flags.EventuallyTimeout).Should(ContainSubstring("group ui-template is
updated"))
rootCmd.SetArgs([]string{"property", "schema", "create", "-a", addr,
"-f", "-"})
createPropertySchema := func() string {
@@ -1002,18 +1007,24 @@ tags:
}
return buf.String()
}
- Eventually(createPropertySchema,
flags.EventuallyTimeout).Should(ContainSubstring("property schema
ui-template.service is created"))
+ Eventually(createPropertySchema, flags.EventuallyTimeout).Should(
+ Or(ContainSubstring("property schema ui-template.service is
created"), ContainSubstring("already exists")))
+ rootCmd.SetArgs([]string{"property", "schema", "update", "-a", addr,
"-f", "-"})
+ Eventually(createPropertySchema,
flags.EventuallyTimeout).Should(ContainSubstring("property schema
ui-template.service is updated"))
}
func applyData(rootCmd *cobra.Command, addr, data string, created bool,
tagsNum int) {
rootCmd.SetArgs([]string{"property", "data", "apply", "-a", addr, "-f",
"-"})
- rootCmd.SetIn(strings.NewReader(data))
var buf bytes.Buffer
- rootCmd.SetOut(&buf)
- rootCmd.SetErr(&buf)
-
- err := rootCmd.Execute()
- Expect(err).NotTo(HaveOccurred())
+ // Right after a node starts, the property shard topology may not be
ready
+ // yet ("failed to get group copies"), so retry until the apply goes
through.
+ Eventually(func() error {
+ rootCmd.SetIn(strings.NewReader(data))
+ buf.Reset()
+ rootCmd.SetOut(&buf)
+ rootCmd.SetErr(&buf)
+ return rootCmd.Execute()
+ }, flags.EventuallyTimeout).Should(Succeed())
out := buf.String()
GinkgoWriter.Println(out)
Expect(out).To(ContainSubstring(fmt.Sprintf("created: %t", created)))
@@ -1036,7 +1047,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..bdf0a2ad6
--- /dev/null
+++ b/pkg/test/eventually.go
@@ -0,0 +1,151 @@
+// 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 (
+ "reflect"
+ "sync/atomic"
+ "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
+// runs and reuses the polling interval, if any; its length adapts to how the
+// Eventually phase converged — flags.QuickStabilityTimeout after a first-try
+// success, flags.StabilityTimeout after a retried convergence (see Should).
+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. It returns true only
when
+// both the Eventually and the Consistently assertion pass, matching the bool
+// return of gomega's Should. optionalDescription is forwarded to both
assertions.
+//
+// The Consistently phase always runs, but its window is adaptive. When the
+// Eventually phase succeeds on its very first evaluation, the state was
already
+// settled before the assertion began, so a short flags.QuickStabilityTimeout
+// window (still a dozen re-samples) suffices to reject a one-off lucky match.
+// Only assertions that had to retry — proof of an in-flight async process that
+// could still regress — pay the full flags.StabilityTimeout window. This keeps
+// the regression guard everywhere without charging every static assertion in
+// the large query tables for a full window.
+func (a ECAssertion) Should(matcher gomegatypes.GomegaMatcher,
optionalDescription ...interface{}) bool {
+ probe, attempts := countEvaluations(a.actual)
+ if !gomega.EventuallyWithOffset(1, probe,
a.eventuallyIntervals()...).Should(matcher, optionalDescription...) {
+ return false
+ }
+ return gomega.ConsistentlyWithOffset(1, a.actual,
a.consistentlyIntervals(atomic.LoadInt32(attempts))...).Should(matcher,
optionalDescription...)
+}
+
+// ShouldNot asserts the matcher is avoided and then kept avoided. It returns
true
+// only when both assertions pass, matching the bool return of gomega's
ShouldNot.
+// optionalDescription is forwarded to both assertions. The stability window is
+// adaptive — see Should.
+func (a ECAssertion) ShouldNot(matcher gomegatypes.GomegaMatcher,
optionalDescription ...interface{}) bool {
+ probe, attempts := countEvaluations(a.actual)
+ if !gomega.EventuallyWithOffset(1, probe,
a.eventuallyIntervals()...).ShouldNot(matcher, optionalDescription...) {
+ return false
+ }
+ return gomega.ConsistentlyWithOffset(1, a.actual,
a.consistentlyIntervals(atomic.LoadInt32(attempts))...).ShouldNot(matcher,
optionalDescription...)
+}
+
+// countEvaluations wraps a func-typed actual so every gomega poll increments a
+// counter, letting the caller tell first-try convergence from retried
+// convergence. Non-func actuals (plain values, channels) are returned as-is
+// with the counter pre-set past 1, so they conservatively keep the full
+// stability window.
+func countEvaluations(actual interface{}) (interface{}, *int32) {
+ attempts := new(int32)
+ v := reflect.ValueOf(actual)
+ if v.Kind() != reflect.Func {
+ *attempts = 2
+ return actual, attempts
+ }
+ probe := reflect.MakeFunc(v.Type(), func(args []reflect.Value)
[]reflect.Value {
+ atomic.AddInt32(attempts, 1)
+ return v.Call(args)
+ }).Interface()
+ return probe, attempts
+}
+
+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(eventuallyAttempts int32)
[]interface{} {
+ window := flags.StabilityTimeout
+ if eventuallyAttempts == 1 {
+ window = flags.QuickStabilityTimeout
+ }
+ out := []interface{}{window}
+ if a.polling != nil {
+ out = append(out, a.polling)
+ }
+ return out
+}
diff --git a/pkg/test/eventually_test.go b/pkg/test/eventually_test.go
new file mode 100644
index 000000000..e25dfa23e
--- /dev/null
+++ b/pkg/test/eventually_test.go
@@ -0,0 +1,62 @@
+// 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_test
+
+import (
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/onsi/gomega"
+
+ "github.com/apache/skywalking-banyandb/pkg/test"
+ "github.com/apache/skywalking-banyandb/pkg/test/flags"
+)
+
+func TestEventuallyConsistentlyUsesQuickWindowOnFirstTryConvergence(t
*testing.T) {
+ gomega.RegisterTestingT(t)
+ var calls int32
+ start := time.Now()
+ test.EventuallyConsistently(func() bool {
+ atomic.AddInt32(&calls, 1)
+ return true
+ }).Should(gomega.BeTrue())
+ elapsed := time.Since(start)
+ if got := atomic.LoadInt32(&calls); got < 2 {
+ t.Fatalf("the stability window must still re-sample after a
first-try convergence, got %d evaluations", got)
+ }
+ if elapsed < flags.QuickStabilityTimeout {
+ t.Fatalf("first-try convergence must hold the quick window %v,
took %v", flags.QuickStabilityTimeout, elapsed)
+ }
+ if elapsed >= flags.StabilityTimeout {
+ t.Fatalf("first-try convergence should pay the quick window,
not the full one, took %v", elapsed)
+ }
+}
+
+func TestEventuallyConsistentlyHoldsWindowAfterRetriedConvergence(t
*testing.T) {
+ gomega.RegisterTestingT(t)
+ var calls int32
+ start := time.Now()
+ test.EventuallyConsistently(func() bool {
+ return atomic.AddInt32(&calls, 1) >= 3
+ }).Should(gomega.BeTrue())
+ elapsed := time.Since(start)
+ if elapsed < flags.StabilityTimeout {
+ t.Fatalf("retried convergence must hold the full stability
window %v, took %v", flags.StabilityTimeout, elapsed)
+ }
+}
diff --git a/pkg/test/flags/flags.go b/pkg/test/flags/flags.go
index eb306f97b..958aa521e 100644
--- a/pkg/test/flags/flags.go
+++ b/pkg/test/flags/flags.go
@@ -30,6 +30,10 @@ var (
consistentlyTimeout string
+ stabilityTimeout string
+
+ quickStabilityTimeout string
+
requireKTM string
// EventuallyTimeout is the timeout of async time cases execution.
@@ -38,6 +42,17 @@ var (
// ConsistentlyTimeout is the timeout of async time cases execution.
ConsistentlyTimeout time.Duration
+ // StabilityTimeout is the Consistently window
test.EventuallyConsistently
+ // appends after an assertion that had to retry before converging.
+ StabilityTimeout time.Duration
+
+ // QuickStabilityTimeout is the Consistently window appended after an
+ // assertion that converged on its first evaluation. It is intentionally
+ // short — the window runs after every assertion, so its length
multiplies
+ // across whole integration suites — yet still re-samples the state a
dozen
+ // times to reject one-off lucky matches.
+ QuickStabilityTimeout time.Duration
+
// NeverTimeout is the timeout of async time cases execution.
NeverTimeout time.Duration
@@ -58,6 +73,12 @@ func init() {
if consistentlyTimeout == "" {
consistentlyTimeout = "5s"
}
+ if stabilityTimeout == "" {
+ stabilityTimeout = "2s"
+ }
+ if quickStabilityTimeout == "" {
+ quickStabilityTimeout = "200ms"
+ }
if requireKTM == "" {
requireKTM = "false"
}
@@ -81,4 +102,14 @@ func init() {
panic(err)
}
ConsistentlyTimeout = d
+ d, err = time.ParseDuration(stabilityTimeout)
+ if err != nil {
+ panic(err)
+ }
+ StabilityTimeout = d
+ d, err = time.ParseDuration(quickStabilityTimeout)
+ if err != nil {
+ panic(err)
+ }
+ QuickStabilityTimeout = d
}
diff --git a/test/cases/lifecycle/lifecycle.go
b/test/cases/lifecycle/lifecycle.go
index 11147bd2e..97254fbe8 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,9 +966,13 @@ 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 {
+ // A per-poll context keeps the
Eventually+Consistently polling from
+ // draining the It-level write context's fixed
budget.
+ qCtx, qCancel :=
context.WithTimeout(context.Background(), 10*time.Second)
+ defer qCancel()
var qErr error
- resp, qErr = queryClient.Query(ctx, req)
+ resp, qErr = queryClient.Query(qCtx, req)
if qErr != nil {
return qErr
}
@@ -1254,9 +1259,11 @@ 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 {
+ qCtx, qCancel :=
context.WithTimeout(context.Background(), 10*time.Second)
+ defer qCancel()
var qErr error
- resp, qErr = queryClient.Query(ctx, req)
+ resp, qErr = queryClient.Query(qCtx, req)
if qErr != nil {
return qErr
}
@@ -1426,9 +1433,11 @@ 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 {
+ qCtx, qCancel :=
context.WithTimeout(context.Background(), 10*time.Second)
+ defer qCancel()
var qErr error
- resp, qErr = queryClient.Query(ctx, req)
+ resp, qErr = queryClient.Query(qCtx, req)
if qErr != nil {
return qErr
}
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/data/data.go b/test/cases/property/data/data.go
index 1844e656b..865f075e8 100644
--- a/test/cases/property/data/data.go
+++ b/test/cases/property/data/data.go
@@ -91,9 +91,14 @@ func verifyWithContext(ctx context.Context, innerGm
gm.Gomega, sharedContext hel
helpers.UnmarshalYAML(w, want)
innerGm.Expect(resp.GetProperties()).To(gm.HaveLen(len(want.GetProperties())),
query.String())
if query.OrderBy == nil {
+ // The server returns unordered scans in nondeterministic
order, so both
+ // sides must be sorted before the order-sensitive cmp.Equal
below.
slices.SortFunc(want.GetProperties(), func(a, b
*propertyv1.Property) int {
return strings.Compare(a.Id, b.Id)
})
+ slices.SortFunc(resp.GetProperties(), func(a, b
*propertyv1.Property) int {
+ return strings.Compare(a.Id, b.Id)
+ })
}
success := innerGm.Expect(cmp.Equal(resp, want,
protocmp.IgnoreUnknown(),
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/cmd/generate/seed.go
b/test/cases/trace/cmd/generate/seed.go
index dda4d1779..9c9ae3b74 100644
--- a/test/cases/trace/cmd/generate/seed.go
+++ b/test/cases/trace/cmd/generate/seed.go
@@ -127,10 +127,10 @@ func swTrace() Trace {
row("trace_001", int64(1), "api_service", "api_instance_1",
"/api/v1/users", int64(750), "span_001_4", "trace_001_span_4"),
row("trace_001", int64(0), "api_service", "api_instance_2",
"/api/v1/profile", int64(320), "span_001_5", "trace_001_span_5"),
row("trace_002", int64(1), "auth_service", "auth_instance_1",
"/auth/login", int64(450), "span_002_3", "trace_002_span_3"),
- row("trace_002", int64(0), "auth_service", "auth_instance_2",
"/auth/validate", int64(180), "span_002_4", "trace_002_span_4"),
+ row("trace_002", int64(0), "auth_service", "auth_instance_2",
"/auth/validate", int64(185), "span_002_4", "trace_002_span_4"),
row("trace_006", int64(1), "payment_service",
"payment_instance_1", "/payment/process", int64(1190), "span_006_1",
"trace_006_span_1"),
row("trace_006", int64(0), "payment_service",
"payment_instance_2", "/payment/verify", int64(280), "span_006_2",
"trace_006_span_2"),
- row("trace_006", int64(0), "payment_service",
"payment_instance_3", "/payment/refund", int64(150), "span_006_3",
"trace_006_span_3"),
+ row("trace_006", int64(0), "payment_service",
"payment_instance_3", "/payment/refund", int64(155), "span_006_3",
"trace_006_span_3"),
row("trace_007", int64(1), "notification_service",
"notification_instance_1", "/notification/send", int64(650), "span_007_1",
"trace_007_span_1"),
row("trace_007", int64(0), "notification_service",
"notification_instance_2", "/notification/email", int64(420), "span_007_2",
"trace_007_span_2"),
row("trace_008", int64(1), "database_service",
"database_instance_1", "/db/query", int64(890), "span_008_1",
"trace_008_span_1"),
diff --git a/test/cases/trace/data/testdata/sw_mixed_traces.json
b/test/cases/trace/data/testdata/sw_mixed_traces.json
index 3728fd93d..3c0f497e9 100644
--- a/test/cases/trace/data/testdata/sw_mixed_traces.json
+++ b/test/cases/trace/data/testdata/sw_mixed_traces.json
@@ -148,7 +148,7 @@
},
{
"int": {
- "value": 180
+ "value": 185
}
},
{
@@ -268,7 +268,7 @@
},
{
"int": {
- "value": 150
+ "value": 155
}
},
{
diff --git
a/test/cases/trace/data/want/gen_feat_order_duration_asc_limit2_offset1_proj_explicit_1.yml
b/test/cases/trace/data/want/gen_feat_order_duration_asc_limit2_offset1_proj_explicit_1.yml
index 719d71cc3..d8f364847 100644
---
a/test/cases/trace/data/want/gen_feat_order_duration_asc_limit2_offset1_proj_explicit_1.yml
+++
b/test/cases/trace/data/want/gen_feat_order_duration_asc_limit2_offset1_proj_explicit_1.yml
@@ -17,64 +17,64 @@
traces:
- spans:
- - span: trace_003_span_1
- spanId: span_003_1
+ - span: trace_006_span_1
+ spanId: span_006_1
tags:
- key: service_id
value:
str:
- value: webapp_service
+ value: payment_service
- key: duration
value:
int:
- value: "1200"
+ value: "1190"
- key: trace_id
value:
str:
- value: trace_003
+ value: trace_006
- key: span_id
value:
str:
- value: span_003_1
- - span: trace_003_span_2
- spanId: span_003_2
+ value: span_006_1
+ - span: trace_006_span_2
+ spanId: span_006_2
tags:
- key: service_id
value:
str:
- value: webapp_service
+ value: payment_service
- key: duration
value:
int:
- value: "150"
+ value: "280"
- key: trace_id
value:
str:
- value: trace_003
+ value: trace_006
- key: span_id
value:
str:
- value: span_003_2
- - span: trace_003_span_3
- spanId: span_003_3
+ value: span_006_2
+ - span: trace_006_span_3
+ spanId: span_006_3
tags:
- key: service_id
value:
str:
- value: webapp_service
+ value: payment_service
- key: duration
value:
int:
- value: "400"
+ value: "155"
- key: trace_id
value:
str:
- value: trace_003
+ value: trace_006
- key: span_id
value:
str:
- value: span_003_3
- traceId: trace_003
+ value: span_006_3
+ traceId: trace_006
- spans:
- span: trace_005_span_1
spanId: span_005_1
diff --git
a/test/cases/trace/data/want/gen_feat_order_duration_asc_limit5_proj_explicit_6.yml
b/test/cases/trace/data/want/gen_feat_order_duration_asc_limit5_proj_explicit_6.yml
index 9ef561847..4fdb08fb3 100644
---
a/test/cases/trace/data/want/gen_feat_order_duration_asc_limit5_proj_explicit_6.yml
+++
b/test/cases/trace/data/want/gen_feat_order_duration_asc_limit5_proj_explicit_6.yml
@@ -17,123 +17,123 @@
traces:
- spans:
- - span: trace_006_span_1
- spanId: span_006_1
+ - span: trace_003_span_1
+ spanId: span_003_1
tags:
- key: service_id
value:
str:
- value: payment_service
+ value: webapp_service
- key: duration
value:
int:
- value: "1190"
+ value: "1200"
- key: trace_id
value:
str:
- value: trace_006
+ value: trace_003
- key: span_id
value:
str:
- value: span_006_1
- - span: trace_006_span_2
- spanId: span_006_2
+ value: span_003_1
+ - span: trace_003_span_2
+ spanId: span_003_2
tags:
- key: service_id
value:
str:
- value: payment_service
+ value: webapp_service
- key: duration
value:
int:
- value: "280"
+ value: "150"
- key: trace_id
value:
str:
- value: trace_006
+ value: trace_003
- key: span_id
value:
str:
- value: span_006_2
- - span: trace_006_span_3
- spanId: span_006_3
+ value: span_003_2
+ - span: trace_003_span_3
+ spanId: span_003_3
tags:
- key: service_id
value:
str:
- value: payment_service
+ value: webapp_service
- key: duration
value:
int:
- value: "150"
+ value: "400"
- key: trace_id
value:
str:
- value: trace_006
+ value: trace_003
- key: span_id
value:
str:
- value: span_006_3
- traceId: trace_006
+ value: span_003_3
+ traceId: trace_003
- spans:
- - span: trace_003_span_1
- spanId: span_003_1
+ - span: trace_006_span_1
+ spanId: span_006_1
tags:
- key: service_id
value:
str:
- value: webapp_service
+ value: payment_service
- key: duration
value:
int:
- value: "1200"
+ value: "1190"
- key: trace_id
value:
str:
- value: trace_003
+ value: trace_006
- key: span_id
value:
str:
- value: span_003_1
- - span: trace_003_span_2
- spanId: span_003_2
+ value: span_006_1
+ - span: trace_006_span_2
+ spanId: span_006_2
tags:
- key: service_id
value:
str:
- value: webapp_service
+ value: payment_service
- key: duration
value:
int:
- value: "150"
+ value: "280"
- key: trace_id
value:
str:
- value: trace_003
+ value: trace_006
- key: span_id
value:
str:
- value: span_003_2
- - span: trace_003_span_3
- spanId: span_003_3
+ value: span_006_2
+ - span: trace_006_span_3
+ spanId: span_006_3
tags:
- key: service_id
value:
str:
- value: webapp_service
+ value: payment_service
- key: duration
value:
int:
- value: "400"
+ value: "155"
- key: trace_id
value:
str:
- value: trace_003
+ value: trace_006
- key: span_id
value:
str:
- value: span_003_3
- traceId: trace_003
+ value: span_006_3
+ traceId: trace_006
- spans:
- span: trace_005_span_1
spanId: span_005_1
@@ -280,7 +280,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
diff --git
a/test/cases/trace/data/want/gen_feat_order_duration_desc_proj_explicit_3.yml
b/test/cases/trace/data/want/gen_feat_order_duration_desc_proj_explicit_3.yml
index 922c88b1b..025ab7a45 100644
---
a/test/cases/trace/data/want/gen_feat_order_duration_desc_proj_explicit_3.yml
+++
b/test/cases/trace/data/want/gen_feat_order_duration_desc_proj_explicit_3.yml
@@ -183,7 +183,7 @@ traces:
- key: duration
value:
int:
- value: "150"
+ value: "155"
- key: trace_id
value:
str:
@@ -436,7 +436,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
diff --git
a/test/cases/trace/data/want/gen_feat_order_timestamp_asc_filter_state_eq_limit5_offset1_proj_explicit_8.yml
b/test/cases/trace/data/want/gen_feat_order_timestamp_asc_filter_state_eq_limit5_offset1_proj_explicit_8.yml
index 384b64fe8..68429d7bf 100644
---
a/test/cases/trace/data/want/gen_feat_order_timestamp_asc_filter_state_eq_limit5_offset1_proj_explicit_8.yml
+++
b/test/cases/trace/data/want/gen_feat_order_timestamp_asc_filter_state_eq_limit5_offset1_proj_explicit_8.yml
@@ -84,7 +84,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
@@ -301,7 +301,7 @@ traces:
- key: duration
value:
int:
- value: "150"
+ value: "155"
- key: trace_id
value:
str:
diff --git a/test/cases/trace/data/want/gen_leaf_eq_state.yml
b/test/cases/trace/data/want/gen_leaf_eq_state.yml
index 05443de1c..2489d5fd4 100644
--- a/test/cases/trace/data/want/gen_leaf_eq_state.yml
+++ b/test/cases/trace/data/want/gen_leaf_eq_state.yml
@@ -164,7 +164,7 @@ traces:
- key: duration
value:
int:
- value: "150"
+ value: "155"
- key: trace_id
value:
str:
@@ -242,7 +242,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
diff --git a/test/cases/trace/data/want/gen_leaf_ge_duration.yml
b/test/cases/trace/data/want/gen_leaf_ge_duration.yml
index 05443de1c..2489d5fd4 100644
--- a/test/cases/trace/data/want/gen_leaf_ge_duration.yml
+++ b/test/cases/trace/data/want/gen_leaf_ge_duration.yml
@@ -164,7 +164,7 @@ traces:
- key: duration
value:
int:
- value: "150"
+ value: "155"
- key: trace_id
value:
str:
@@ -242,7 +242,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
diff --git a/test/cases/trace/data/want/gen_leaf_ge_state.yml
b/test/cases/trace/data/want/gen_leaf_ge_state.yml
index 05443de1c..2489d5fd4 100644
--- a/test/cases/trace/data/want/gen_leaf_ge_state.yml
+++ b/test/cases/trace/data/want/gen_leaf_ge_state.yml
@@ -164,7 +164,7 @@ traces:
- key: duration
value:
int:
- value: "150"
+ value: "155"
- key: trace_id
value:
str:
@@ -242,7 +242,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
diff --git a/test/cases/trace/data/want/gen_leaf_gt_duration.yml
b/test/cases/trace/data/want/gen_leaf_gt_duration.yml
index 05443de1c..2489d5fd4 100644
--- a/test/cases/trace/data/want/gen_leaf_gt_duration.yml
+++ b/test/cases/trace/data/want/gen_leaf_gt_duration.yml
@@ -164,7 +164,7 @@ traces:
- key: duration
value:
int:
- value: "150"
+ value: "155"
- key: trace_id
value:
str:
@@ -242,7 +242,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
diff --git a/test/cases/trace/data/want/gen_leaf_gt_state.yml
b/test/cases/trace/data/want/gen_leaf_gt_state.yml
index 05443de1c..2489d5fd4 100644
--- a/test/cases/trace/data/want/gen_leaf_gt_state.yml
+++ b/test/cases/trace/data/want/gen_leaf_gt_state.yml
@@ -164,7 +164,7 @@ traces:
- key: duration
value:
int:
- value: "150"
+ value: "155"
- key: trace_id
value:
str:
@@ -242,7 +242,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
diff --git a/test/cases/trace/data/want/gen_leaf_in_service_id.yml
b/test/cases/trace/data/want/gen_leaf_in_service_id.yml
index 195753b8b..2bd1d9b87 100644
--- a/test/cases/trace/data/want/gen_leaf_in_service_id.yml
+++ b/test/cases/trace/data/want/gen_leaf_in_service_id.yml
@@ -84,7 +84,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
diff --git a/test/cases/trace/data/want/gen_leaf_in_trace_id.yml
b/test/cases/trace/data/want/gen_leaf_in_trace_id.yml
index 435dfdc12..cd3ef8d63 100644
--- a/test/cases/trace/data/want/gen_leaf_in_trace_id.yml
+++ b/test/cases/trace/data/want/gen_leaf_in_trace_id.yml
@@ -181,7 +181,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
diff --git a/test/cases/trace/data/want/gen_leaf_le_duration.yml
b/test/cases/trace/data/want/gen_leaf_le_duration.yml
index 05443de1c..2489d5fd4 100644
--- a/test/cases/trace/data/want/gen_leaf_le_duration.yml
+++ b/test/cases/trace/data/want/gen_leaf_le_duration.yml
@@ -164,7 +164,7 @@ traces:
- key: duration
value:
int:
- value: "150"
+ value: "155"
- key: trace_id
value:
str:
@@ -242,7 +242,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
diff --git a/test/cases/trace/data/want/gen_leaf_le_state.yml
b/test/cases/trace/data/want/gen_leaf_le_state.yml
index 05443de1c..2489d5fd4 100644
--- a/test/cases/trace/data/want/gen_leaf_le_state.yml
+++ b/test/cases/trace/data/want/gen_leaf_le_state.yml
@@ -164,7 +164,7 @@ traces:
- key: duration
value:
int:
- value: "150"
+ value: "155"
- key: trace_id
value:
str:
@@ -242,7 +242,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
diff --git a/test/cases/trace/data/want/gen_leaf_lt_duration.yml
b/test/cases/trace/data/want/gen_leaf_lt_duration.yml
index 05443de1c..2489d5fd4 100644
--- a/test/cases/trace/data/want/gen_leaf_lt_duration.yml
+++ b/test/cases/trace/data/want/gen_leaf_lt_duration.yml
@@ -164,7 +164,7 @@ traces:
- key: duration
value:
int:
- value: "150"
+ value: "155"
- key: trace_id
value:
str:
@@ -242,7 +242,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
diff --git a/test/cases/trace/data/want/gen_leaf_lt_state.yml
b/test/cases/trace/data/want/gen_leaf_lt_state.yml
index 05443de1c..2489d5fd4 100644
--- a/test/cases/trace/data/want/gen_leaf_lt_state.yml
+++ b/test/cases/trace/data/want/gen_leaf_lt_state.yml
@@ -164,7 +164,7 @@ traces:
- key: duration
value:
int:
- value: "150"
+ value: "155"
- key: trace_id
value:
str:
@@ -242,7 +242,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
diff --git a/test/cases/trace/data/want/gen_leaf_ne_service_id.yml
b/test/cases/trace/data/want/gen_leaf_ne_service_id.yml
index 05443de1c..2489d5fd4 100644
--- a/test/cases/trace/data/want/gen_leaf_ne_service_id.yml
+++ b/test/cases/trace/data/want/gen_leaf_ne_service_id.yml
@@ -164,7 +164,7 @@ traces:
- key: duration
value:
int:
- value: "150"
+ value: "155"
- key: trace_id
value:
str:
@@ -242,7 +242,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
diff --git a/test/cases/trace/data/want/gen_leaf_ne_state.yml
b/test/cases/trace/data/want/gen_leaf_ne_state.yml
index 05443de1c..2489d5fd4 100644
--- a/test/cases/trace/data/want/gen_leaf_ne_state.yml
+++ b/test/cases/trace/data/want/gen_leaf_ne_state.yml
@@ -164,7 +164,7 @@ traces:
- key: duration
value:
int:
- value: "150"
+ value: "155"
- key: trace_id
value:
str:
@@ -242,7 +242,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
diff --git a/test/cases/trace/data/want/gen_leaf_ne_trace_id.yml
b/test/cases/trace/data/want/gen_leaf_ne_trace_id.yml
index 05443de1c..2489d5fd4 100644
--- a/test/cases/trace/data/want/gen_leaf_ne_trace_id.yml
+++ b/test/cases/trace/data/want/gen_leaf_ne_trace_id.yml
@@ -164,7 +164,7 @@ traces:
- key: duration
value:
int:
- value: "150"
+ value: "155"
- key: trace_id
value:
str:
@@ -242,7 +242,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
diff --git a/test/cases/trace/data/want/gen_leaf_not_in_service_id.yml
b/test/cases/trace/data/want/gen_leaf_not_in_service_id.yml
index 05443de1c..2489d5fd4 100644
--- a/test/cases/trace/data/want/gen_leaf_not_in_service_id.yml
+++ b/test/cases/trace/data/want/gen_leaf_not_in_service_id.yml
@@ -164,7 +164,7 @@ traces:
- key: duration
value:
int:
- value: "150"
+ value: "155"
- key: trace_id
value:
str:
@@ -242,7 +242,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
diff --git a/test/cases/trace/data/want/gen_leaf_not_in_trace_id.yml
b/test/cases/trace/data/want/gen_leaf_not_in_trace_id.yml
index 05443de1c..2489d5fd4 100644
--- a/test/cases/trace/data/want/gen_leaf_not_in_trace_id.yml
+++ b/test/cases/trace/data/want/gen_leaf_not_in_trace_id.yml
@@ -164,7 +164,7 @@ traces:
- key: duration
value:
int:
- value: "150"
+ value: "155"
- key: trace_id
value:
str:
@@ -242,7 +242,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
diff --git a/test/cases/trace/data/want/gen_tree_depth5_deep_and.yml
b/test/cases/trace/data/want/gen_tree_depth5_deep_and.yml
index 05443de1c..2489d5fd4 100644
--- a/test/cases/trace/data/want/gen_tree_depth5_deep_and.yml
+++ b/test/cases/trace/data/want/gen_tree_depth5_deep_and.yml
@@ -164,7 +164,7 @@ traces:
- key: duration
value:
int:
- value: "150"
+ value: "155"
- key: trace_id
value:
str:
@@ -242,7 +242,7 @@ traces:
- key: duration
value:
int:
- value: "180"
+ value: "185"
- key: trace_id
value:
str:
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,
