edge case

Author: func25

app/vlstorage/netselect/async_tasks.go

@@ -42,6 +42,7 @@ func (s *Storage) ListAsyncTasks(ctx context.Context) ([]logstorage.AsyncTaskInf
 	for _, ts := range results {
 		all = append(all, ts...)
 	}
+
 	return all, nil
 }
 

lib/logstorage/async_task_info.go

@@ -62,6 +62,35 @@ func (s *Storage) ListAsyncTasks() []AsyncTaskInfo {
 	}()
 
 	var out []AsyncTaskInfo
+	seqToIdx := make(map[uint64]int)
+	// helper to merge two tasks with the same Seq
+	merge := func(dst *AsyncTaskInfo, src *AsyncTaskInfo) {
+		// status: error > pending > success
+		prioritize := func(s asyncTaskStatus) int {
+			switch s {
+			case taskError:
+				return 3
+			case taskPending:
+				return 2
+			case taskSuccess:
+				return 1
+			}
+			return 0
+		}
+		if prioritize(src.Status) > prioritize(dst.Status) {
+			dst.Status = src.Status
+			dst.Error = src.Error
+		}
+		// earliest created time
+		if dst.CreatedTime == 0 || (src.CreatedTime != 0 && src.CreatedTime < dst.CreatedTime) {
+			dst.CreatedTime = src.CreatedTime
+		}
+		// latest done time
+		if src.DoneTime > dst.DoneTime {
+			dst.DoneTime = src.DoneTime
+		}
+	}
+
 	for _, p := range pws {
 		a := p.pt.ats
 
@@ -70,7 +99,13 @@ func (s *Storage) ListAsyncTasks() []AsyncTaskInfo {
 		a.mu.Unlock()
 
 		for _, t := range tasks {
-			out = append(out, asyncTaskToInfo(t))
+			info := asyncTaskToInfo(t)
+			if idx, ok := seqToIdx[info.Seq]; ok {
+				merge(&out[idx], &info)
+				continue
+			}
+			seqToIdx[info.Seq] = len(out)
+			out = append(out, info)
 		}
 	}
 

lib/logstorage/storage.go

@@ -824,12 +824,14 @@ func ValidateDeleteQuery(q *Query) error {
 	}
 
 	minTS, maxTS := q.GetFilterTimeRange()
-	now := int64(fasttime.UnixTimestamp())
-
-	// vlselect already adds a timestamp, but with floor(now).
-	// To avoid adding _time twice, use seconds to compare with now().
-	if maxTS/1e9 > now {
-		q.AddTimeFilter(minTS, now*1e9)
+	now := int64(fasttime.UnixTimestamp() * 1e9)
+
+	// vlselect already adds a timestamp floor(now).
+	// vlstorage parses the time by adding +0.999, so we need to use now+1 to avoid
+	// duplicating the addition of _time filters.
+	if maxTS > now+int64(time.Second) {
+		logger.Infof("DEBUG: maxTS=%d > now=%d, adjusting maxTS to now", maxTS, now)
+		q.AddTimeFilter(minTS, now)
 	}
 
 	return nil