[ntcore] Special-case default timestamps (#5003)

Previously, a setDefault() on the server could override a client doing a
real set() if the time offset between client and server was negative,
resulting in a negative timestamp from the client.  This is a not
uncommon situation with robot code, as the robot code always starts at
time 0, so any clients that set values earlier (in real time) would have
negative timestamps.

Also improve special casing of 0 in the transmit side to make sure a
normal timestamp will never get sent as 0.

ntcore/src/main/native/cpp/LocalStorage.cpp

@@ -501,7 +501,8 @@ bool LSImpl::SetValue(TopicData* topic, const Value& value,
   if (topic->type != NT_UNASSIGNED && topic->type != value.type()) {
     return false;
   }
-  if (!topic->lastValue || value.time() >= topic->lastValue.time()) {
+  if (!topic->lastValue || topic->lastValue.time() == 0 ||
+      value.time() >= topic->lastValue.time()) {
     // TODO: notify option even if older value
     topic->type = value.type();
     topic->lastValue = value;

ntcore/src/main/native/cpp/net/ClientImpl.cpp

@@ -268,6 +268,10 @@ void CImpl::SendValues(uint64_t curTimeMs, bool flush) {
         int64_t time = val.time();
         if (time != 0) {
           time += m_serverTimeOffsetUs;
+          // make sure resultant time isn't exactly 0
+          if (time == 0) {
+            time = 1;
+          }
         }
         WireEncodeBinary(writer.Add(), Handle{pub->handle}.GetIndex(), time,
                          val);

ntcore/src/main/native/cpp/net/ServerImpl.cpp

@@ -2129,7 +2129,7 @@ void SImpl::SetFlags(ClientData* client, TopicData* topic, unsigned int flags) {
 void SImpl::SetValue(ClientData* client, TopicData* topic, const Value& value) {
   // update retained value if from same client or timestamp newer
   if (!topic->lastValue || topic->lastValueClient == client ||
-      value.time() >= topic->lastValue.time()) {
+      topic->lastValue.time() == 0 || value.time() >= topic->lastValue.time()) {
     DEBUG4("updating '{}' last value (time was {} is {})", topic->name,
            topic->lastValue.time(), value.time());
     topic->lastValue = value;

ntcore/src/test/native/cpp/net/ServerImplTest.cpp

@@ -95,12 +95,20 @@ static std::string EncodeText(const T& msgs) {
 }
 
 template <typename T>
-static std::vector<uint8_t> EncodeBinary(const T& msgs) {
+static std::vector<uint8_t> EncodeServerBinary(const T& msgs) {
   std::vector<uint8_t> data;
   wpi::raw_uvector_ostream os{data};
   for (auto&& msg : msgs) {
-    if (auto m = std::get_if<net::ServerValueMsg>(&msg.contents)) {
-      net::WireEncodeBinary(os, m->topic, m->value.time(), m->value);
+    if constexpr (std::is_same_v<typename T::value_type, net::ServerMessage>) {
+      if (auto m = std::get_if<net::ServerValueMsg>(&msg.contents)) {
+        net::WireEncodeBinary(os, m->topic, m->value.time(), m->value);
+      }
+    } else if constexpr (std::is_same_v<typename T::value_type,
+                                        net::ClientMessage>) {
+      if (auto m = std::get_if<net::ClientValueMsg>(&msg.contents)) {
+        net::WireEncodeBinary(os, Handle{m->pubHandle}.GetIndex(),
+                              m->value.time(), m->value);
+      }
     }
   }
   return data;
@@ -231,8 +239,9 @@ TEST_F(ServerImplTest, ClientSubTopicOnlyThenValue) {
       std::vector<net::ServerMessage> smsgs;
       smsgs.emplace_back(net::ServerMessage{
           net::ServerValueMsg{3, Value::MakeDouble(1.0, 10)}});
-      EXPECT_CALL(wire,
-                  Binary(wpi::SpanEq(EncodeBinary(smsgs))));  // SendValues()
+      EXPECT_CALL(
+          wire,
+          Binary(wpi::SpanEq(EncodeServerBinary(smsgs))));  // SendValues()
     }
     EXPECT_CALL(wire, Flush());  // SendValues()
   }
@@ -266,4 +275,59 @@ TEST_F(ServerImplTest, ClientSubTopicOnlyThenValue) {
   server.SendValues(id, 200);
 }
 
+TEST_F(ServerImplTest, ZeroTimestampNegativeTime) {
+  // publish before client connect
+  server.SetLocal(&local);
+  NT_Publisher pubHandle = nt::Handle{0, 1, nt::Handle::kPublisher};
+  NT_Topic topicHandle = nt::Handle{0, 1, nt::Handle::kTopic};
+  NT_Subscriber subHandle = nt::Handle{0, 1, nt::Handle::kSubscriber};
+  Value defaultValue = Value::MakeDouble(1.0, 10);
+  defaultValue.SetTime(0);
+  defaultValue.SetServerTime(0);
+  Value value = Value::MakeDouble(5, -10);
+  {
+    ::testing::InSequence seq;
+    EXPECT_CALL(local, NetworkAnnounce("test", "double", wpi::json::object(),
+                                       pubHandle))
+        .WillOnce(Return(topicHandle));
+    EXPECT_CALL(local, NetworkSetValue(topicHandle, defaultValue));
+    EXPECT_CALL(local, NetworkSetValue(topicHandle, value));
+  }
+
+  {
+    std::vector<net::ClientMessage> msgs;
+    msgs.emplace_back(net::ClientMessage{net::PublishMsg{
+        pubHandle, topicHandle, "test", "double", wpi::json::object(), {}}});
+    msgs.emplace_back(
+        net::ClientMessage{net::ClientValueMsg{pubHandle, defaultValue}});
+    msgs.emplace_back(
+        net::ClientMessage{net::SubscribeMsg{subHandle, {"test"}, {}}});
+    server.HandleLocal(msgs);
+  }
+
+  // client connect; it should get already-published topic as soon as it
+  // subscribes
+  ::testing::StrictMock<net::MockWireConnection> wire;
+  MockSetPeriodicFunc setPeriodic;
+  {
+    ::testing::InSequence seq;
+    EXPECT_CALL(wire, Flush());  // AddClient()
+  }
+  auto [name, id] = server.AddClient("test", "connInfo", false, wire,
+                                     setPeriodic.AsStdFunction());
+
+  // publish and send non-default value with negative time offset
+  {
+    NT_Subscriber pubHandle2 = nt::Handle{0, 2, nt::Handle::kPublisher};
+    std::vector<net::ClientMessage> msgs;
+    msgs.emplace_back(net::ClientMessage{net::PublishMsg{
+        pubHandle2, topicHandle, "test", "double", wpi::json::object(), {}}});
+    server.ProcessIncomingText(id, EncodeText(msgs));
+    msgs.clear();
+    msgs.emplace_back(
+        net::ClientMessage{net::ClientValueMsg{pubHandle2, value}});
+    server.ProcessIncomingBinary(id, EncodeServerBinary(msgs));
+  }
+}
+
 }  // namespace nt