Add tests

libs/postgres_ffi/src/xlog_utils.rs

@@ -502,4 +502,25 @@ pub fn encode_logical_message(prefix: &str, message: &str) -> Vec<u8> {
     wal
 }
 
-// If you need to craft WAL and write tests for this module, put it at wal_craft crate.
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_ts_conversion() {
+        let now = SystemTime::now();
+        let round_trip = from_pg_timestamp(to_pg_timestamp(now));
+
+        assert_eq!(
+            now.elapsed().unwrap().as_micros(),
+            round_trip.elapsed().unwrap().as_micros()
+        );
+
+        let now_pg = get_current_timestamp();
+        let round_trip_pg = to_pg_timestamp(from_pg_timestamp(now_pg));
+
+        assert_eq!(now_pg, round_trip_pg);
+    }
+
+    // If you need to craft WAL and write tests for this module, put it at wal_craft crate.
+}

pageserver/src/http/routes.rs

@@ -2,6 +2,7 @@
 //! Management HTTP API
 //!
 use std::collections::HashMap;
+use std::str::FromStr;
 use std::sync::Arc;
 
 use anyhow::{anyhow, Context, Result};
@@ -514,7 +515,7 @@ async fn get_time_range_of_lsn_handler(
     let timeline_id: TimelineId = parse_request_param(&request, "timeline_id")?;
 
     let lsn_str = must_get_query_param(&request, "lsn")?;
-    let lsn = Lsn::from_hex(&lsn_str)
+    let lsn = Lsn::from_str(&lsn_str)
         .with_context(|| format!("Invalid LSN: {lsn_str:?}"))
         .map_err(ApiError::BadRequest)?;
 

test_runner/fixtures/pageserver/http.py

@@ -453,6 +453,17 @@ def timeline_get_lsn_by_timestamp(
         res_json = res.json()
         return res_json
 
+    def timeline_get_time_range_of_lsn(
+        self, tenant_id: TenantId, timeline_id: TimelineId, lsn: Lsn
+    ):
+        log.info(f"Requesting time range of lsn {lsn}, tenant {tenant_id}, timeline {timeline_id}")
+        res = self.get(
+            f"http://localhost:{self.port}/v1/tenant/{tenant_id}/timeline/{timeline_id}/get_time_range_of_lsn?lsn={lsn}",
+        )
+        self.verbose_error(res)
+        res_json = res.json()
+        return res_json
+
     def timeline_checkpoint(self, tenant_id: TenantId, timeline_id: TimelineId):
         self.is_testing_enabled_or_skip()
 

test_runner/regress/test_lsn_mapping.py

@@ -1,7 +1,9 @@
-from datetime import timedelta
+from datetime import datetime, timedelta, timezone
 
 from fixtures.log_helper import log
 from fixtures.neon_fixtures import NeonEnvBuilder, wait_for_last_flush_lsn
+from fixtures.pageserver.http import PageserverApiException
+from fixtures.types import Lsn
 from fixtures.utils import query_scalar
 
 
@@ -67,3 +69,99 @@ def test_lsn_mapping(neon_env_builder: NeonEnvBuilder):
             assert endpoint_here.safe_psql("SELECT max(x) FROM foo")[0][0] == i
 
             endpoint_here.stop_and_destroy()
+
+
+# Test pageserver get_time_range_of_lsn API
+def test_ts_of_lsn_api(neon_env_builder: NeonEnvBuilder):
+    env = neon_env_builder.init_start()
+
+    new_timeline_id = env.neon_cli.create_branch("test_ts_of_lsn_api")
+    endpoint_main = env.endpoints.create_start("test_ts_of_lsn_api")
+    log.info("postgres is running on 'test_ts_of_lsn_api' branch")
+
+    cur = endpoint_main.connect().cursor()
+    # Create table, and insert rows, each in a separate transaction
+    # Disable synchronous_commit to make this initialization go faster.
+    #
+    # Each row contains current insert LSN and the current timestamp, when
+    # the row was inserted.
+    cur.execute("SET synchronous_commit=off")
+    cur.execute("CREATE TABLE foo (x integer)")
+    tbl = []
+    for i in range(1000):
+        cur.execute(f"INSERT INTO foo VALUES({i})")
+        # Get the timestamp at UTC
+        after_timestamp = query_scalar(cur, "SELECT clock_timestamp()").replace(tzinfo=timezone.utc)
+        after_lsn = query_scalar(cur, "SELECT pg_current_wal_lsn()")
+        tbl.append([i, after_timestamp, after_lsn])
+
+    # Execute one more transaction with synchronous_commit enabled, to flush
+    # all the previous transactions
+    cur.execute("INSERT INTO foo VALUES (-1)")
+
+    # Wait until WAL is received by pageserver
+    wait_for_last_flush_lsn(env, endpoint_main, env.initial_tenant, new_timeline_id)
+
+    last_flush_lsn = Lsn(query_scalar(cur, "SELECT pg_current_wal_flush_lsn()"))
+
+    with env.pageserver.http_client() as client:
+        # Check edge cases: lsn larger than the last flush lsn
+        probe_lsn = Lsn(int(last_flush_lsn) * 20 + 80_000)
+        result = client.timeline_get_time_range_of_lsn(
+            env.initial_tenant,
+            new_timeline_id,
+            probe_lsn,
+        )
+        # assert result == "future"
+
+        # lsn of zero
+        try:
+            probe_lsn = Lsn(0)
+            result = client.timeline_get_time_range_of_lsn(
+                env.initial_tenant,
+                new_timeline_id,
+                probe_lsn,
+            )
+            # There should always be an error here.
+            AssertionError()
+        except PageserverApiException as error:
+            assert error.status_code == 500
+            assert str(error) == "Invalid LSN"
+            env.pageserver.allowed_errors.append(".*Invalid LSN.*")
+
+        # small lsn before initdb_lsn
+        try:
+            probe_lsn = Lsn(64)
+            result = client.timeline_get_time_range_of_lsn(
+                env.initial_tenant,
+                new_timeline_id,
+                probe_lsn,
+            )
+            # There should always be an error here.
+            AssertionError()
+        except PageserverApiException as error:
+            assert error.status_code == 500
+            assert str(error).startswith("could not find data for key")
+            env.pageserver.allowed_errors.append(".*could not find data for key.*")
+
+        # Probe a bunch of timestamps in the valid range
+        for i in range(1, len(tbl), 100):
+            after_timestamp = tbl[i][1]
+            after_lsn = tbl[i][2]
+            result = client.timeline_get_time_range_of_lsn(
+                env.initial_tenant,
+                new_timeline_id,
+                after_lsn,
+            )
+            log.info("result: %s, after_ts: %s", result, after_timestamp)
+
+            # Ensure that empty is not set or set to false
+            assert ("empty" not in result) or not (result["empty"])
+
+            min = datetime.fromisoformat(result["min"]).replace(tzinfo=timezone.utc)
+            max = datetime.fromisoformat(result["max"]).replace(tzinfo=timezone.utc)
+            assert min <= max, "min smaller than max"
+            assert max < after_timestamp, "after_timestamp after max"
+            if i > 1:
+                before_timestamp = tbl[i - 100][1]
+                assert min >= before_timestamp, "before_timestamp before min"