Copilot commented on code in PR #1145:
URL:
https://github.com/apache/skywalking-banyandb/pull/1145#discussion_r3324611032
##########
banyand/backup/lifecycle/segment_boundary_utils.go:
##########
@@ -24,17 +24,16 @@ import (
"github.com/apache/skywalking-banyandb/pkg/timestamp"
)
-func calculateTargetSegments(partMinTS, partMaxTS int64, targetInterval
storage.IntervalRule) []time.Time {
- // Use time.Local so sender's bucket math stays on the same per-timezone
- // grid as the receiver, which sees ctx.MinTimestamp as time.Local too.
- minTime := time.Unix(0, partMinTS)
- maxTime := time.Unix(0, partMaxTS)
-
+// calculateTargetSegments returns every target bucket overlapping the
half-open
+// source range [start, end). Every migration caller passes the source
segment's
+// own boundaries (segment/shard time range), where end is exclusive, so the
+// strict current.Before(end) guard naturally excludes the empty next bucket
+// when the source ends exactly on a target boundary. Using time.Local keeps
the
+// sender's bucket math on the same per-timezone grid as the receiver, which
+// sees ctx.MinTimestamp as time.Local too.
+func calculateTargetSegments(start, end time.Time, targetInterval
storage.IntervalRule) []time.Time {
var targetSegments []time.Time
- // current starts at the bucket containing minTime, so segmentEnd is
- // always > minTime and the loop guard already excludes anything past
- // maxTime - every iteration produces a real overlap.
- for current := targetInterval.Standard(minTime);
!current.After(maxTime); current = targetInterval.NextTime(current) {
+ for current := targetInterval.Standard(start); current.Before(end);
current = targetInterval.NextTime(current) {
targetSegments = append(targetSegments, current)
}
Review Comment:
calculateTargetSegments' doc says the bucket math should be on the
time.Local grid, but the implementation currently uses start/end's Location
implicitly (storage.IntervalRule.Standard anchors at 1970-01-01 in
t.Location()). Converting start/end to time.Local inside the helper makes the
behavior match the comment and avoids timezone-dependent behavior if a caller
passes a UTC time.Time.
##########
test/cases/lifecycle/lifecycle.go:
##########
@@ -398,3 +487,687 @@ func asInt(v interface{}) int {
}
return 0
}
+
+// crossSegmentTimestamps returns three probe timestamps shared by the
+// cross-segment suites: single sits in a source segment fully inside one
target
+// segment, while left/right sit in a source segment that straddles a target
+// boundary (left and right of the crossing). The picked source day N satisfies
+// N%6==2 (LCM of the 2-day source and 3-day target grids), shifted >=8 local
+// days back so it clears the 5-day source TTL.
+func crossSegmentTimestamps() (single, left, right time.Time) {
+ nowLocal := time.Now().Local().Truncate(24 * time.Hour)
+ nowDay := nowLocal.Unix() / (24 * 3600)
+ crossSrcDayStart := nowDay - 8
+ for crossSrcDayStart%6 != 2 {
+ crossSrcDayStart--
+ }
+ crossSrcStart := nowLocal.AddDate(0, 0, int(crossSrcDayStart-nowDay))
+ return crossSrcStart.Add(-12 * time.Hour), crossSrcStart.Add(12 *
time.Hour), crossSrcStart.Add(36 * time.Hour)
+}
+
+// runLifecycleMigration runs a single hot->warm lifecycle migration, pointing
+// every root path at the shared source dir and writing its report to
reportDir.
+func runLifecycleMigration(progressFile, reportDir string) {
+ lifecycleCmd := lifecycle.NewCommand()
+ args := []string{
+ "--grpc-addr", SharedContext.DataAddr,
+ "--stream-root-path", SharedContext.SrcDir,
+ "--measure-root-path", SharedContext.SrcDir,
+ "--trace-root-path", SharedContext.SrcDir,
+ "--progress-file", progressFile,
+ "--report-dir", reportDir,
+ }
+ args = append(args, SharedContext.MetadataFlags...)
+ lifecycleCmd.SetArgs(args)
+ gomega.Expect(lifecycleCmd.Execute()).To(gomega.Succeed())
+}
+
+// drainWriteAcks closes a client-streaming write and fails the spec if any
+// server-side ack reports a non-success status, surfacing per-row rejections
+// that a bare CloseSend would swallow.
+func drainWriteAcks[R interface{ GetStatus() string }](recv func() (R, error),
closeSend func() error) {
+ ackErrs := make(chan error, 8)
+ go func() {
+ defer close(ackErrs)
+ for {
+ resp, recvErr := recv()
+ if recvErr != nil {
+ return
+ }
+ if status := resp.GetStatus(); status != "" && status
!= "STATUS_SUCCEED" {
+ ackErrs <- fmt.Errorf("write rejected: %s",
status)
+ }
+ }
+ }()
+ gomega.Expect(closeSend()).To(gomega.Succeed())
+ for ackErr := range ackErrs {
+ gomega.Expect(ackErr).NotTo(gomega.HaveOccurred(), "write ack
must succeed")
+ }
+}
Review Comment:
drainWriteAcks currently (1) ignores Recv() errors (it just returns), so
non-EOF stream failures can be missed, and (2) can deadlock if more than 8
non-success acks are produced before CloseSend returns (bounded channel fills
and blocks the recv goroutine). A simple synchronous receive loop after
CloseSend avoids both issues and fails the spec on any non-success status or
non-EOF error.
##########
test/cases/lifecycle/lifecycle.go:
##########
@@ -19,17 +19,27 @@
package lifecycle_test
import (
+ "context"
"encoding/json"
+ "fmt"
"io/fs"
"os"
Review Comment:
The new drainWriteAcks helper should verify the Recv() terminal error
(io.EOF) and surface non-EOF stream errors. That requires importing io;
otherwise Recv errors can be silently swallowed and the test can pass even if
the write stream fails server-side.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
