Disambiguate zero-offset log cursors

2026-11-08 01:16:47 +08:00 · 2026-06-15 15:50:03 +08:00
parent 917cec3bf6
commit db3fdea4a1
2 changed files with 134 additions and 5 deletions
@@ -527,6 +527,7 @@ type logCursor struct {
 	Offset          int64  `json:"offset"`
 	Size            int64  `json:"size"`
 	ModTime         int64  `json:"modTime"`
+	ModTimeUnixNano int64  `json:"modTimeUnixNano,omitempty"`
 	LatestTimestamp int64  `json:"latestTimestamp"`
 	Fingerprint     string `json:"fingerprint"`
 }
@@ -808,6 +809,7 @@ func locateLogCursorFile(files []string, cursor logCursor) (int, bool, error) {
 	for i := range files {
 		nameToIndex[filepath.Base(files[i])] = i
 	}
+	deferEmptyMainMatch := false
 	if index, ok := nameToIndex[cursor.File]; ok {
 		matches, truncated, errMatch := logFileMatchesCursor(files[index], cursor)
 		if errMatch != nil {
@@ -817,17 +819,24 @@ func locateLogCursorFile(files []string, cursor logCursor) (int, bool, error) {
 			return 0, false, errMatch
 		}
 		if matches && !truncated {
-			return index, true, nil
+			if shouldDeferEmptyMainCursorToRotated(files, cursor) {
+				deferEmptyMainMatch = true
+			} else {
+				return index, true, nil
+			}
 		}
 	}

-	if cursor.File != defaultLogFileName || cursor.Offset == 0 {
+	if cursor.File != defaultLogFileName || (cursor.Offset == 0 && cursor.Size == 0 && !deferEmptyMainMatch) {
 		return 0, false, nil
 	}
-	for i := range files {
+	for i := len(files) - 1; i >= 0; i-- {
 		if filepath.Base(files[i]) == defaultLogFileName {
 			continue
 		}
+		if cursor.Offset == 0 && cursor.Size == 0 && !logFileChangedAfterCursor(files[i], cursor) {
+			continue
+		}
 		matches, truncated, errMatch := logFileMatchesCursor(files[i], cursor)
 		if errMatch != nil {
 			if errors.Is(errMatch, os.ErrNotExist) {
@@ -845,6 +854,29 @@ func locateLogCursorFile(files []string, cursor logCursor) (int, bool, error) {
 	return 0, false, nil
 }

+func shouldDeferEmptyMainCursorToRotated(files []string, cursor logCursor) bool {
+	if cursor.File != defaultLogFileName || cursor.Offset != 0 || cursor.Size != 0 {
+		return false
+	}
+	for i := range files {
+		if filepath.Base(files[i]) == defaultLogFileName {
+			continue
+		}
+		if logFileChangedAfterCursor(files[i], cursor) {
+			return true
+		}
+	}
+	return false
+}
+
+func logFileChangedAfterCursor(path string, cursor logCursor) bool {
+	info, errStat := os.Stat(path)
+	if errStat != nil || info.IsDir() || info.Size() == 0 {
+		return false
+	}
+	return info.ModTime().UnixNano() > cursorModTimeUnixNano(cursor)
+}
+
 func logFileMatchesCursor(path string, cursor logCursor) (bool, bool, error) {
 	info, errStat := os.Stat(path)
 	if errStat != nil {
@@ -856,7 +888,11 @@ func logFileMatchesCursor(path string, cursor logCursor) (bool, bool, error) {
 	if info.Size() < cursor.Offset {
 		return false, true, nil
 	}
-	fingerprint, errFingerprint := logFileFingerprint(path, cursor.Offset)
+	boundary := cursorFingerprintBoundary(cursor)
+	if info.Size() < boundary {
+		return false, true, nil
+	}
+	fingerprint, errFingerprint := logFileFingerprint(path, boundary)
 	if errFingerprint != nil {
 		return false, false, errFingerprint
 	}
@@ -953,7 +989,11 @@ func newLogCursor(path string, offset, latest int64) (string, error) {
 	if offset < 0 || offset > info.Size() {
 		return "", fmt.Errorf("invalid cursor offset")
 	}
-	fingerprint, errFingerprint := logFileFingerprint(path, offset)
+	fingerprintCursor := logCursor{
+		Offset: offset,
+		Size:   info.Size(),
+	}
+	fingerprint, errFingerprint := logFileFingerprint(path, cursorFingerprintBoundary(fingerprintCursor))
 	if errFingerprint != nil {
 		return "", errFingerprint
 	}
@@ -963,11 +1003,26 @@ func newLogCursor(path string, offset, latest int64) (string, error) {
 		Offset:          offset,
 		Size:            info.Size(),
 		ModTime:         info.ModTime().Unix(),
+		ModTimeUnixNano: info.ModTime().UnixNano(),
 		LatestTimestamp: latest,
 		Fingerprint:     fingerprint,
 	})
 }

+func cursorFingerprintBoundary(cursor logCursor) int64 {
+	if cursor.Offset == 0 && cursor.Size > 0 {
+		return cursor.Size
+	}
+	return cursor.Offset
+}
+
+func cursorModTimeUnixNano(cursor logCursor) int64 {
+	if cursor.ModTimeUnixNano > 0 {
+		return cursor.ModTimeUnixNano
+	}
+	return cursor.ModTime * int64(time.Second)
+}
+
 func logFileFingerprint(path string, boundary int64) (string, error) {
 	if boundary < 0 {
 		return "", fmt.Errorf("invalid fingerprint boundary")
@@ -382,6 +382,80 @@ func TestGetLogsCursorReadsRotatedFileWhenNewMainIsSmaller(t *testing.T) {
 	}
 }

+func TestGetLogsZeroOffsetCursorWithPartialLineReadsAcrossRotation(t *testing.T) {
+	dir := t.TempDir()
+	writeMainLog(t, dir, "partial")
+	initial := performGetLogs(t, newLogsTestHandler(dir, true), "/v0/management/logs?limit=1")
+	if initial.NextCursor == "" {
+		t.Fatal("initial next-cursor is empty")
+	}
+	cursor, errCursor := decodeLogCursor(initial.NextCursor)
+	if errCursor != nil {
+		t.Fatalf("decode initial cursor: %v", errCursor)
+	}
+	if cursor.Offset != 0 || cursor.Size == 0 {
+		t.Fatalf("cursor offset/size = %d/%d, want zero offset with partial size", cursor.Offset, cursor.Size)
+	}
+
+	appendMainLog(t, dir, " complete\n")
+	if err := os.Rename(filepath.Join(dir, defaultLogFileName), filepath.Join(dir, defaultLogFileName+".1")); err != nil {
+		t.Fatalf("rotate main log: %v", err)
+	}
+	writeMainLog(t, dir, "new\n")
+
+	resp := performGetLogs(t, newLogsTestHandler(dir, true), "/v0/management/logs?cursor="+url.QueryEscape(initial.NextCursor)+"&limit=10")
+	wantLines := []string{"partial complete", "new"}
+	if !reflect.DeepEqual(resp.Lines, wantLines) {
+		t.Fatalf("lines = %#v, want %#v", resp.Lines, wantLines)
+	}
+	if resp.CursorReset {
+		t.Fatal("cursor-reset = true, want false")
+	}
+}
+
+func TestGetLogsZeroOffsetCursorWithEmptyFileReadsAcrossRotation(t *testing.T) {
+	dir := t.TempDir()
+	writeMainLog(t, dir, "")
+	initial := performGetLogs(t, newLogsTestHandler(dir, true), "/v0/management/logs?limit=1")
+	if initial.NextCursor == "" {
+		t.Fatal("initial next-cursor is empty")
+	}
+	cursor, errCursor := decodeLogCursor(initial.NextCursor)
+	if errCursor != nil {
+		t.Fatalf("decode initial cursor: %v", errCursor)
+	}
+	if cursor.Offset != 0 || cursor.Size != 0 {
+		t.Fatalf("cursor offset/size = %d/%d, want empty zero offset", cursor.Offset, cursor.Size)
+	}
+
+	appendMainLog(t, dir, "first\n")
+	mainPath := filepath.Join(dir, defaultLogFileName)
+	nextModTime := time.Unix(0, cursorModTimeUnixNano(cursor)+int64(time.Second))
+	if err := os.Chtimes(mainPath, nextModTime, nextModTime); err != nil {
+		t.Fatalf("update main log mtime: %v", err)
+	}
+	if err := os.Rename(mainPath, filepath.Join(dir, defaultLogFileName+".1")); err != nil {
+		t.Fatalf("rotate main log: %v", err)
+	}
+	writeMainLog(t, dir, "second\n")
+
+	resp := performGetLogs(t, newLogsTestHandler(dir, true), "/v0/management/logs?cursor="+url.QueryEscape(initial.NextCursor)+"&limit=1")
+	if !reflect.DeepEqual(resp.Lines, []string{"first"}) {
+		t.Fatalf("lines = %#v, want first rotated line", resp.Lines)
+	}
+	if resp.CursorReset {
+		t.Fatal("cursor-reset = true, want false")
+	}
+
+	next := performGetLogs(t, newLogsTestHandler(dir, true), "/v0/management/logs?cursor="+url.QueryEscape(resp.NextCursor)+"&limit=1")
+	if !reflect.DeepEqual(next.Lines, []string{"second"}) {
+		t.Fatalf("next lines = %#v, want second main line", next.Lines)
+	}
+	if next.CursorReset {
+		t.Fatal("next cursor-reset = true, want false")
+	}
+}
+
 func TestGetLogsInvalidCursorResetsToTail(t *testing.T) {
 	dir := t.TempDir()
 	lines := []string{