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 <[email protected]> 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,
