feat(topic): add Topic.latest() for a single-read value + timestamp

Checking whether a published value is stale today means two calls:

    long stamp = topic.latestPublishNanos();
    T v = topic.latestValueOr(null);
    long age = System.nanoTime() - stamp;

Both accessors are individually correct, but a publish can land between
them, leaving the caller holding a value from one publish and a timestamp
from another. Reading the timestamp first makes that safe in one
direction -- the age can only be over-reported, never under-reported --
but over-reporting still discards a fresh value as stale, by up to a full
publish interval.

How often that happens depends on how long the two reads take. Normally
they are nanoseconds apart and a publish landing between them is
vanishingly rare. It stops being rare when the reader is preempted
between the two calls by a GC pause or the scheduler, since the window
becomes milliseconds. This library configures a small young gen, so young
GCs are part of the deployment rather than a hypothetical.

`latest()` returns the value and its timestamp from one snapshot read, so
the pair provably comes from a single publish and the window does not
exist:

    Optional<Topic.Latest<T>> snap = topic.latest();
    T v = snap.map(Topic.Latest::value).orElse(default);
    long age = snap.map(Topic.Latest::ageNanos).orElse(Long.MAX_VALUE);

Additive only. No existing signature or behaviour changes, and callers
that never check an age can keep using `latestValueOr` -- same single
volatile read.

`Topic.Latest` becomes public with `value()`, `publishNanos()` and
`ageNanos()`. `ageNanos()` is the age of *this* value computed from its
own stamp, which also removes the pre-publish `0` sentinel that the
two-call form has to guard against by hand.

`latestPublishNanos()` still returns 0 before the first publish, so the
guard stays necessary for anyone composing it by hand. It is documented as
such in the javadoc, the CHANGELOG and the topics guide.

The concurrency test's reader now uses `latest()`, which is the stronger
property: no straddle is representable, so the "timestamp first"
ordering rule has nothing left to guard. That test previously carried a
cross-check between `latest()` being empty and `latestPublishNanos()`
being nonzero; it is gone, because a publish can legitimately land
between those two reads and a nonzero stamp there says nothing about
consistency.

Suite: 61 tests, 0 failures.

Author: IamCoder18

CHANGELOG.md

@@ -42,7 +42,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   **Read the timestamp before the value** when you use the two together as a
   staleness check. Each accessor now returns a self-consistent pair, but two separate
   calls can still straddle a publish, so the ordering rule remains — it is now
-  documented on the public accessors and in the topics guide:
+  documented on the public accessors and in the topics guide. Prefer the new
+  `Topic.latest()` when you need both: it returns the value and its timestamp from one
+  snapshot read, so the two provably come from the same publish and the ordering rule
+  does not apply.
 
   ```java
   long stamp = topic.latestPublishNanos();  // first

src/main/java/com/aaravlabs/synapse/Topic.java

@@ -46,15 +46,50 @@ public final class Topic<T> {
         this.boxedType = box(type);
     }
 
-    /** One publish's value and the {@link System#nanoTime()} at which it was recorded. */
-    private static final class Latest<T> {
-        final T value;
-        final long publishNanos;
+    /**
+     * One publish's value and the {@link System#nanoTime()} at which it was recorded.
+     *
+     * <p>Immutable. Returned by {@link Topic#latest()} so a caller can read the value and
+     * its timestamp without risking a publish landing between two calls.
+     *
+     * @param <T> the message type carried by this topic
+     */
+    public static final class Latest<T> {
+
+        private final T value;
+        private final long publishNanos;
 
         Latest(T value, long publishNanos) {
             this.value = value;
             this.publishNanos = publishNanos;
         }
+
+        /** @return the published value */
+        public T value() {
+            return value;
+        }
+
+        /**
+         * @return the {@link System#nanoTime()} at which {@link #value()} was recorded, on
+         *         the monotonic clock (arbitrary origin — only differences are meaningful)
+         */
+        public long publishNanos() {
+            return publishNanos;
+        }
+
+        /**
+         * @return nanos elapsed since {@link #publishNanos()}, on the monotonic clock.
+         *         Unlike {@code System.nanoTime() - publishNanos()}, this is always the age
+         *         of <i>this</i> value and needs no {@code 0} sentinel handling.
+         */
+        public long ageNanos() {
+            return System.nanoTime() - publishNanos;
+        }
+
+        @Override
+        public String toString() {
+            return "Latest[" + value + " @" + publishNanos + "]";
+        }
     }
 
     /**
@@ -74,9 +109,9 @@ public Class<T> type() {
     /**
      * The most recently published value, or empty if nothing has been published yet.
      *
-     * <p><b>Reading a value together with its age:</b> each accessor reads a
-     * self-consistent pair, but two separate calls can still straddle a publish. Sample
-     * the timestamp <i>first</i>, which can only over-report the value's age:
+     * <p><b>Reading a value together with its age:</b> prefer {@link #latest()}, which
+     * returns both from a single snapshot read. Composing them from two accessors works
+     * only if you sample the timestamp <i>first</i>, which can only over-report the age:
      *
      * <pre>{@code
      * long stamp = topic.latestPublishNanos();  // first; 0 means nothing published yet
@@ -88,7 +123,7 @@ public Class<T> type() {
      * {@link #latestPublishNanos()} is {@code 0}, and subtracting it would yield the raw
      * {@link System#nanoTime()} reading — seconds to days — rather than an age.
      *
-     * Reading the value first is the unsafe order: a publish landing between the two
+     * <p>Reading the value first is the unsafe order: a publish landing between the two
      * calls pairs the older value with the newer timestamp, so an age check on that
      * pair passes even though the value is stale.
      *
@@ -110,7 +145,7 @@ public Optional<T> latestValue() {
      */
     public long latestPublishNanos() {
         Latest<T> snap = latest;
-        return snap == null ? 0L : snap.publishNanos;
+        return snap == null ? 0L : snap.publishNanos();
     }
 
     /**
@@ -165,7 +200,39 @@ boolean acceptsType(Class<?> other) {
      */
     public T latestValueOr(T defaultValue) {
         Latest<T> snap = latest;
-        return snap == null ? defaultValue : snap.value;
+        return snap == null ? defaultValue : snap.value();
+    }
+
+    /**
+     * The most recent value and the {@link System#nanoTime()} at which it was published,
+     * as one consistent pair.
+     *
+     * <p><b>Prefer this over {@link #latestValue()} plus {@link #latestPublishNanos()}
+     * when you are checking whether a value is stale.</b> The two accessors each read the
+     * snapshot correctly, but a caller making two calls can straddle a publish between
+     * them and end up holding a value from one publish and a timestamp from another:
+     *
+     * <pre>{@code
+     * long stamp = topic.latestPublishNanos();   // publish A
+     *                                                  ...publish B lands...
+     * T v = topic.latestValueOr(null);             // publish B's value, with A's stamp
+     * long age = System.nanoTime() - stamp;        // over-reports by a full publish interval
+     * }</pre>
+     *
+     * <p>Reading the timestamp first makes that safe in one direction — the age can only
+     * be over-reported, never under-reported — but over-reporting still costs you: a
+     * fresh value gets rejected as stale. How often that happens depends on how long the
+     * two reads take. Normally they are nanoseconds apart and a publish landing between
+     * them is vanishingly rare. It stops being rare when the reader is preempted between
+     * the two calls by a GC pause or the scheduler, since the window becomes
+     * milliseconds. This accessor removes the window entirely.
+     *
+     * <p>Returns empty before the first publish.
+     *
+     * @return an {@link Optional} holding the latest value and its publish timestamp
+     */
+    public Optional<Latest<T>> latest() {
+        return Optional.ofNullable(latest);
     }
 
     /** Treat primitive {@code double.class} and wrapper {@code Double.class} as the same type. */

src/test/java/com/aaravlabs/synapse/TopicLockFreeTest.java

@@ -78,6 +78,9 @@ void concurrentPublishAndReadNeverLosesOrTearsTheLatestValue() throws Exception
         // not evidence that the code is correct. What this test does establish is that no
         // reader ever observes a value outside the published range, and that the final
         // state is some publisher's last value.
+        // Timestamp and value from ONE snapshot read, so there is no window for a
+        // publish to land in between. This is the property the two-call accessors cannot
+        // offer, and the reason latest() exists.
         final int publishers = 4;
         final int perPublisher = 25_000;
         final int ids = publishers * perPublisher;
@@ -106,32 +109,37 @@ void concurrentPublishAndReadNeverLosesOrTearsTheLatestValue() throws Exception
         Thread reader = new Thread(() -> {
             try {
                 while (readersRunning.get()) {
-                    // Timestamp FIRST, then value. On this order a publish landing
-                    // between the two reads can only make the reported age too large,
-                    // never too small, so a staleness check can never pass a stale value.
-                    long ts = topic.latestPublishNanos();
-                    Integer v = topic.latestValueOr(null);
-                    if (v == null) {
-                        if (ts != 0L) {
-                            failure.compareAndSet(null, new AssertionError(
-                                    "timestamp recorded with no value: " + ts));
+                    // ONE snapshot read, so the value and the timestamp provably come
+                    // from the same publish. There is no window for a publish to land in,
+                    // so this cannot over-report age the way two calls can.
+                    Topic.Latest<Integer> snap = topic.latest().orElse(null);
+                    if (snap == null) {
+                        // Nothing published yet. There is deliberately no cross-check
+                        // against latestPublishNanos() here: a publish can land between
+                        // the two calls, so a nonzero stamp here would be that publish,
+                        // not an inconsistency.
+                    } else {
+                        Integer v = snap.value();
+                        if (v < 0 || v >= ids) {
+                            failure.compareAndSet(null,
+                                    new AssertionError("torn/garbage latest value: " + v));
                             return;
                         }
-                    } else if (v < 0 || v >= ids) {
-                        failure.compareAndSet(null,
-                                new AssertionError("torn/garbage latest value: " + v));
-                        return;
-                    } else {
-                        // Only a timestamp AFTER the window ends is a tear: the reader
-                        // would be holding an older value beside a newer timestamp. A
-                        // timestamp before the window is the safe, expected straddle --
-                        // the reader sampled the timestamp, a publish landed, then it read
-                        // the newer value.
+                        // The pair is self-consistent by construction, so the only way
+                        // to fail is a stamp that predates the value it came with --
+                        // which the publish-window check bounds.
                         long hi = windowHi.get(v);
-                        if (hi != Long.MIN_VALUE && ts > hi) {
+                        if (hi != Long.MIN_VALUE && snap.publishNanos() > hi) {
                             failure.compareAndSet(null, new AssertionError(
-                                    "value " + v + " is older than the timestamp " + ts
-                                            + " the reader sampled; its publish ended at " + hi));
+                                    "value " + v + " paired with stamp " + snap.publishNanos()
+                                            + " later than its publish end " + hi));
+                            return;
+                        }
+                        // ageNanos must agree with the stamp it was taken from.
+                        long age = snap.ageNanos();
+                        if (age <= 0 || age > System.nanoTime() - snap.publishNanos() + 1) {
+                            failure.compareAndSet(null, new AssertionError(
+                                    "implausible age " + age + " for stamp " + snap.publishNanos()));
                             return;
                         }
                     }
@@ -240,6 +248,44 @@ void acceptsValueClassTreatsPrimitivesAndWrappersAsEquivalent() {
         assertFalse(i.acceptsValueClass(double.class), "Double must not match Integer topic");
     }
 
+    @Test
+    void latestReturnsValueAndTimestampFromOneConsistentPublish() throws Exception {
+        // latest() exists so a staleness check reads one snapshot instead of two. It must
+        // be empty before the first publish, agree with both single-field accessors, and
+        // never need a 0 sentinel.
+        Topic<String> t = orchestrator.getOrCreateTopic("snap", String.class);
+        assertFalse(t.latest().isPresent(), "latest() must be empty before the first publish");
+
+        orchestrator.publish("snap", "a");
+        Topic.Latest<String> first = t.latest().orElseThrow();
+        assertEquals("a", first.value());
+        assertEquals(first.publishNanos(), t.latestPublishNanos(),
+                "latest() and latestPublishNanos() must describe the same publish");
+        assertEquals("a", t.latestValueOr(null));
+        assertTrue(first.publishNanos() != 0L,
+                "a published stamp must not be the pre-publish 0 sentinel");
+
+        // ageNanos is the age of THIS value, computed from its own stamp. Taking a fresh
+        // clock reading afterwards must give an equal or LARGER delta, never a smaller
+        // one -- a smaller delta would mean the age came from somewhere other than this
+        // stamp, which is exactly the straddle latest() exists to prevent.
+        long age = first.ageNanos();
+        assertTrue(age >= 0 && age < 1_000_000_000L, "ageNanos out of range: " + age);
+        assertTrue(age <= System.nanoTime() - first.publishNanos(),
+                "ageNanos " + age + " exceeds a delta measured later from the same stamp");
+
+        // A later publish replaces both halves together; there is no observable state in
+        // which one half is from this publish and the other from the previous one.
+        Thread.sleep(2);
+        orchestrator.publish("snap", "b");
+        Topic.Latest<String> second = t.latest().orElseThrow();
+        assertEquals("b", second.value());
+        assertTrue(second.publishNanos() > first.publishNanos(),
+                "a later publish must carry a later stamp");
+        assertTrue(second.ageNanos() < first.ageNanos(),
+                "the newer value must report a younger age");
+    }
+
     @Test
     void acceptsTypeIsEquivalentToTheOldBoxedAssignableFromCheck() {
         // acceptsType() replaced `boxed(t.type()).isAssignableFrom(boxed(type))` at

website/src/content/docs/concepts/topics.mdx

@@ -52,19 +52,16 @@ double fl = latest.orElse(0.0);
 
 Topic<Double> flTopic = orchestrator.findTopic("drive/power/fl", Double.class).orElseThrow();
 
-// Read the timestamp first, then the value. Each accessor is internally consistent,
-// but two separate calls can still straddle a publish, so this order is the only one
-// that cannot pair an older value with a newer timestamp and report a fresh age for a
-// value that is already stale.
-//
-// latestPublishNanos() is 0 until the topic's first publish, so treat 0 as "never
-// published" rather than subtracting it -- that would hand you the raw nanoTime
-// reading, which is seconds to days rather than an age.
-long stamp = flTopic.latestPublishNanos();
-long nanosSinceUpdate = stamp == 0L ? Long.MAX_VALUE : System.nanoTime() - stamp;
-double fl2 = flTopic.latestValueOr(0.0);
+// One snapshot read gives the value and its timestamp together, so the two can never
+// come from different publishes. ageNanos() is the age of this exact value, and needs no
+// special handling before the first publish -- latest() is simply empty then.
+Optional<Topic.Latest<Double>> flSnapshot = flTopic.latest();
+double fl2 = flSnapshot.map(Topic.Latest::value).orElse(0.0);
+long nanosSinceUpdate = flSnapshot.map(Topic.Latest::ageNanos).orElse(Long.MAX_VALUE);
 ```
 
+If you only want the value and never check its age, keep using `latestValueOr(0.0)` — it is the same single volatile read and reads better at a glance.
+
 Fetching the latest value is the right tool inside periodic loops; subscribing is the right tool when the *event* matters, not the current state. See [Subscriptions](/docs/concepts/subscriptions) for the tradeoff.
 
 ## Next steps