Fix expired records not being dropped on receipt

docs/en/changes/changes.md

@@ -58,6 +58,7 @@
 * Fix OAL `>`, `>=`, `<` and `<=` filters on a `long` or `double` source field, and `==` on a `long` one, failing to compile when they compare it with a number, such as `filter(sideCar.internalRequestLatencyNanos >= 0)`. The generated code passed the number as an `int`, and the comparison matchers have no `match(long, int)` or `match(double, int)` overload, as Javassist does not widen an argument. The number is now written in the field's type, `0L` or `0.0`; filters on `int` fields, which already worked, generate the same code.
 * Add `ZIPKIN` to the GenAI evaluation record's `traceRef.type`, for spans received by `receiver-zipkin` and OTLP spans converted with `otlpTraceStorage: zipkin`, so the type names the store holding the judged span: `SKYWALKING_NATIVE`, `OTLP` for OTLP spans stored natively, or `ZIPKIN`. Zipkin and OTLP spans are both addressed by span ID, and one OAP can hold both, so an `OTLP` reference alone could not tell a client which store to read the trace from. The type is decided when the span is received, so records written under different `otlpTraceStorage` settings each name their own store.
 * Fix the Zabbix receiver dropping the sub-second part of agent sample times. Each `agent data` item carries `clock` (Unix seconds) and `ns` (nanoseconds within that second), but the sample time used only `clock * 1000`; it is now `clock * 1000 + ns / 1_000_000`. The sub-second precision can affect values produced by custom Zabbix rules using MAL `rate`, `irate` or `increase`; results from the bundled Zabbix rules and the assignment of samples to minute buckets are unchanged.
+* Fix expired records never being dropped on receipt. `RecordStreamProcessor` decoded a record's second-precision time bucket as a minute bucket, which put the record's time far in the future, so no record was ever treated as expired. It also compared the record's age against `recordDataTTL` converted to microseconds instead of milliseconds.
 
 #### UI
 * Add a Virtual GenAI evaluation-record page and evaluation-score chart in Horizon UI, so operators can inspect evaluation result, level, reason, judge model, timestamp, trace linkage, and the `gen_ai_model_evaluation_score_ppm` trend for evaluated records.

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordStreamProcessor.java

@@ -65,8 +65,8 @@ public static RecordStreamProcessor getInstance() {
     @Override
     public void in(Record record) {
         final var now = System.currentTimeMillis();
-        final var recordTimestamp = TimeBucket.getTimestamp(record.getTimeBucket(), DownSampling.Minute);
-        final var isExpired = now - recordTimestamp > TimeUnit.DAYS.toMicros(recordDataTTL);
+        final var recordTimestamp = TimeBucket.getTimestamp(record.getTimeBucket(), DownSampling.Second);
+        final var isExpired = now - recordTimestamp > TimeUnit.DAYS.toMillis(recordDataTTL);
         if (isExpired && !isTestingTTL) {
             log.debug("Receiving expired record: {}, time: {}, ignored", record.id(), record.getTimeBucket());
             return;

oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/worker/RecordStreamProcessorTest.java

@@ -0,0 +1,77 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.core.analysis.worker;
+
+import java.lang.reflect.Field;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
+import org.apache.skywalking.oap.server.core.analysis.record.Record;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+class RecordStreamProcessorTest {
+    private static final int TTL_DAYS = 3;
+
+    private RecordStreamProcessor processor;
+    private RecordPersistentWorker worker;
+    private Record record;
+
+    @BeforeEach
+    @SuppressWarnings("unchecked")
+    void setUp() throws Exception {
+        processor = new RecordStreamProcessor();
+        processor.setRecordDataTTL(TTL_DAYS);
+        worker = mock(RecordPersistentWorker.class);
+        record = mock(Record.class);
+
+        final Field workersField = RecordStreamProcessor.class.getDeclaredField("workers");
+        workersField.setAccessible(true);
+        ((Map<Class<? extends Record>, RecordPersistentWorker>) workersField.get(processor))
+            .put(record.getClass(), worker);
+    }
+
+    @Test
+    void acceptsRecordWithinTTL() {
+        givenRecordAged(TimeUnit.DAYS.toMillis(TTL_DAYS) - TimeUnit.MINUTES.toMillis(2));
+
+        processor.in(record);
+
+        verify(worker).in(record);
+    }
+
+    @Test
+    void dropsRecordBeyondTTL() {
+        givenRecordAged(TimeUnit.DAYS.toMillis(TTL_DAYS) + TimeUnit.MINUTES.toMillis(2));
+
+        processor.in(record);
+
+        verifyNoInteractions(worker);
+    }
+
+    private void givenRecordAged(final long ageMillis) {
+        when(record.getTimeBucket()).thenReturn(
+            TimeBucket.getRecordTimeBucket(System.currentTimeMillis() - ageMillis));
+    }
+}

test/e2e-v2/cases/storage/banyandb/data-generate/docker-compose.yml

@@ -34,6 +34,10 @@ services:
       service: oap
     environment:
       SW_STORAGE: banyandb
+      # The mockers send records MINUS_DAYS (3) in the past, past the default 3-day recordDataTTL that
+      # RecordStreamProcessor drops on receipt. Only this generator OAP accepts them; the stages case
+      # runs with the default TTL and reads them from the files mounted on its cold node.
+      SW_CORE_RECORD_DATA_TTL: 5
       SW_STORAGE_BANYANDB_RECORDS_TTL_DAYS: 5
       SW_STORAGE_BANYANDB_TRACE_TTL_DAYS: 5
       SW_STORAGE_BANYANDB_ZIPKIN_TRACE_TTL_DAYS: 5