Fix logordering race condition for broken_config test (#627)

.gencode_hash.txt

@@ -4,7 +4,7 @@ e693079943921c88e24140521637c58bc9aadfb9aee96dfba414bbaa7be958d0  gencode/docs/c
 20fc22e10cf8ac92d21e1fb5b59052de11ec5d842d36615ac51815dc12588ed9  gencode/docs/config_mapping.html
 cc4bb19794042f88f3a72b197b4d21736e5cebf6ec47ffac560430e9f580181f  gencode/docs/configuration_endpoint.html
 4da8ddd26aed7428171e1cbc6d134b37e30a958d09ce0f255bbb22b130d26f77  gencode/docs/configuration_execution.html
-65925f1af21c535ad347f6db8a7d7f73df95cdca2435401e227d6710bcb562fa  gencode/docs/configuration_pubber.html
+366b0927ba7cec795ee9c84e04bbe9fddbcb796166f413c12e217ba31dac0a5e  gencode/docs/configuration_pubber.html
 a2851c70bd7c7787f760db9e8b29f8604a606dcdf1d7447da363ddf354ff3669  gencode/docs/envelope.html
 41d26633d8592eabd36cbb8344884cc04ee93bbac942dd73336206d34dad0f93  gencode/docs/event.html
 c392f72cc92525587e8473354f2d8827096c68d2e418ad8f4121136be98aeb52  gencode/docs/event_discovery.html
@@ -92,7 +92,7 @@ fc3a9415c04d8a06954dbdbfdff5d68ab113cce3948532c19df555778ffb04fa  gencode/java/u
 ca2e7566106818ca7e5190c8041eb86f0c9b3251b0bda8c3ea7ce11a0c891a0a  gencode/java/udmi/schema/Position.java
 3df66bb1a37a9e0b2b6cf392f8c64d404a73c83e5e13c02bb4844f09b9a04b70  gencode/java/udmi/schema/Properties.java
 783320700f7e96cf8e421461f14999afca10cd540ca6aadf184add39f0ac048c  gencode/java/udmi/schema/PubberConfiguration.java
-11c0149a3d4ca7bf069b6699b19a26a401bc611fb17c2d8c97f4844cc6054feb  gencode/java/udmi/schema/PubberOptions.java
+95bf2511d6f598bcda4bf70c0181e3b902219128188d482d69d6a7313c587c42  gencode/java/udmi/schema/PubberOptions.java
 4604c0aab7cf63af4523e57a52e971013398adb5558500d7a453db49b8e1995a  gencode/java/udmi/schema/ReflectorConfig.java
 abe99dd74122c186403baa6982300a9d5968f8bbb7a67b1689104111b98f32fb  gencode/java/udmi/schema/ReflectorState.java
 4983cc00c17d11346a0353e7af726cd302e84ff4a1eabb7efcf2f4289b4ba81e  gencode/java/udmi/schema/SequenceValidationState.java
@@ -165,7 +165,7 @@ ee9c02c35438fb7d9aacb15a21ec7b35b533c1000d0bde044ec3923b1fdccca4  gencode/python
 15b349141ebae651c6c3c5c313b197d49c8b2b44e8ff1b0639848ad42e5c4e63  gencode/python/udmi/schema/model_system_hardware.py
 aafe6e70c281152db958adf77a024e3e9fab8293927106297c5ec48c11f54e27  gencode/python/udmi/schema/model_testing.py
 5c50847e136a033ea511209238bb570499b43fbee6189dae06603132dcb9f01f  gencode/python/udmi/schema/model_testing_target.py
-e4ecd1c86af0e5e0c249efc9924c1187fb44d368c99e76c68def5d1c96cf8378  gencode/python/udmi/schema/options_pubber.py
+61149ec70e1e47ba4b6f4c8f450567ff9de15509f4fa7a17b831660e5b63fd02  gencode/python/udmi/schema/options_pubber.py
 6c5f3dd1c5ca9d821e3c48298af118fc7eafd97af9265dfd34b2ed8642efca77  gencode/python/udmi/schema/persistent_device.py
 a58f8c98e837a5b56126ca0f410e02f1e9cfcd80a8cb429e0ef522defab1f690  gencode/python/udmi/schema/properties.py
 7fd3cd24bc9704778eba11696d21c4e88a35aea6e37e468d88666bc40eaafbad  gencode/python/udmi/schema/reflect_config.py

etc/test_itemized.out

@@ -1,3 +1,4 @@
+TEST broken_config configStateDelay RESULT pass system broken_config ALPHA 5 Sequence complete
 TEST endpoint_failure_and_restart RESULT fail endpoint endpoint_failure_and_restart DISABLED 5 timeout waiting for system mode is INITIAL
 TEST valid_serial_no noLastStart RESULT pass system valid_serial_no ALPHA 5 Sequence complete
 TEST writeback_success noWriteback RESULT fail writeback writeback_success ALPHA 5 timeout waiting for point filter_differential_pressure_setpoint to have value_state applied

pubber/src/main/java/daq/pubber/Pubber.java

@@ -2,6 +2,7 @@
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
+import static com.google.common.base.Preconditions.checkState;
 import static com.google.udmi.util.GeneralUtils.deepCopy;
 import static com.google.udmi.util.GeneralUtils.fromJsonFile;
 import static com.google.udmi.util.GeneralUtils.fromJsonString;
@@ -171,6 +172,7 @@ public class Pubber {
   private static final long BYTES_PER_MEGABYTE = 1024 * 1024;
   private static final String CORRUPT_STATE_MESSAGE = "!&*@(!*&@!";
   private static final long INJECT_MESSAGE_DELAY_MS = 2000; // Delay to make sure testing is stable.
+  private static final int FORCED_STATE_TIME_MS = 10000;
   final State deviceState = new State();
   private final File outDir;
   private final ScheduledExecutorService executor = new CatchingScheduledThreadPoolExecutor(1);
@@ -936,13 +938,32 @@ private void publisherHandler(String type, String phase, Throwable cause) {
     publishLogMessage(report);
     // TODO: Replace this with a heap so only the highest-priority status is reported.
     deviceState.system.status = shouldLogLevel(report.level) ? report : null;
-    publishAsynchronousState();
+    publishConfigStateUpdate();
     if (cause != null && configLatch.getCount() > 0) {
       configLatch.countDown();
       warn("Released startup latch because reported error");
     }
   }
 
+  /**
+   * Issue a state update in response to a received config message. This will optionally
+   * add a synthetic delay in so that testing infrastructure can test that related sequence
+   * tests handle this case appropriately.
+   */
+  private void publishConfigStateUpdate() {
+    if (TRUE.equals(configuration.options.configStateDelay)) {
+      delayNextStateUpdate();
+    }
+    publishAsynchronousState();
+  }
+
+  private void delayNextStateUpdate() {
+    // Calculate a synthetic last state time that factors in the optional delay.
+    long syntheticType = System.currentTimeMillis() - STATE_THROTTLE_MS + FORCED_STATE_TIME_MS;
+    // And use the synthetic time iff it's later than the actual last state time.
+    lastStateTimeMs = Math.max(lastStateTimeMs, syntheticType);
+  }
+
   private boolean shouldLogLevel(int level) {
     Integer minLoglevel = deviceConfig.system == null ? null : deviceConfig.system.min_loglevel;
     return level >= (minLoglevel == null ? Level.INFO.value() : minLoglevel);
@@ -986,7 +1007,7 @@ private void configHandler(Config config) {
     } catch (Exception e) {
       publisherConfigLog("apply", e);
     }
-    publishAsynchronousState();
+    publishConfigStateUpdate();
   }
 
   private void processConfigUpdate(Config config) {
@@ -1494,7 +1515,8 @@ private void publishLogMessage(Entry report) {
   private void publishAsynchronousState() {
     if (stateLock.tryAcquire()) {
       try {
-        long delay = lastStateTimeMs + STATE_THROTTLE_MS - System.currentTimeMillis();
+        long soonestAllowedStateUpdate = lastStateTimeMs + STATE_THROTTLE_MS;
+        long delay = soonestAllowedStateUpdate - System.currentTimeMillis();
         debug(String.format("State update defer %dms", delay));
         if (delay > 0) {
           markStateDirty(delay);
@@ -1545,7 +1567,6 @@ private void publishStateMessage(Object stateToSend) {
       warn(String.format("State update delay %dms", delay));
       safeSleep(delay);
     }
-
     lastStateTimeMs = System.currentTimeMillis();
     CountDownLatch latch = new CountDownLatch(1);
     publishDeviceMessage(stateToSend, () -> {

schema/options_pubber.json

@@ -28,6 +28,9 @@
     "extraPoint": {
       "type": "string"
     },
+    "configStateDelay": {
+      "type": "boolean"
+    },
     "missingPoint": {
       "type": "string"
     },

validator/src/main/java/com/google/daq/mqtt/sequencer/SequenceBase.java

@@ -104,6 +104,7 @@ public class SequenceBase {
 
   private static final int FUNCTIONS_VERSION_BETA = Validator.REQUIRED_FUNCTION_VER;
   private static final int FUNCTIONS_VERSION_ALPHA = 6; // Version required for alpha execution.
+  private static final long CONFIG_BARRIER_MS = 1000;
 
   static {
     // Sanity check to make sure ALPHA version is increased if forced by increased BETA.
@@ -723,6 +724,8 @@ private void assertConfigIsNotPending() {
 
   protected void updateConfig(String reason) {
     assertConfigIsNotPending();
+    // Add a forced sleep to make sure second-quantized timestamps are unique.
+    safeSleep(CONFIG_BARRIER_MS);
     updateConfig(SubFolder.SYSTEM, augmentConfig(deviceConfig.system));
     updateConfig(SubFolder.POINTSET, deviceConfig.pointset);
     updateConfig(SubFolder.GATEWAY, deviceConfig.gateway);
@@ -1346,6 +1349,12 @@ protected void checkThatHasInterestingSystemStatus(boolean isInteresting) {
     check.accept("interesting system status", this::hasInterestingSystemStatus);
   }
 
+  protected void untilHasInterestingSystemStatus(boolean isInteresting) {
+    BiConsumer<String, Supplier<Boolean>> until =
+        isInteresting ? this::untilTrue : this::untilFalse;
+    until.accept("interesting system status", this::hasInterestingSystemStatus);
+  }
+
   /**
    * Add a summary of a test, with a simple description of what it's testing.
    */

validator/src/main/java/com/google/daq/mqtt/sequencer/sequences/ConfigSequences.java

@@ -91,7 +91,7 @@ public void broken_config() {
 
     setExtraField("break_json");
     untilLogged(SYSTEM_CONFIG_RECEIVE, SYSTEM_CONFIG_RECEIVE_LEVEL);
-    checkThatHasInterestingSystemStatus(true);
+    untilHasInterestingSystemStatus(true);
     Entry stateStatus = deviceState.system.status;
     info("Error message: " + stateStatus.message);
     debug("Error detail: " + stateStatus.detail);
@@ -106,14 +106,19 @@ public void broken_config() {
     untilLogged(SYSTEM_CONFIG_PARSE, Level.ERROR);
     checkNotLogged(SYSTEM_CONFIG_APPLY, SYSTEM_CONFIG_APPLY_LEVEL);
 
+    // Will restore min_loglevel to the default of INFO.
     resetConfig(); // clears extra_field
+    untilLogged(SYSTEM_CONFIG_RECEIVE, SYSTEM_CONFIG_RECEIVE_LEVEL);
+    untilLogged(SYSTEM_CONFIG_APPLY, SYSTEM_CONFIG_APPLY_LEVEL);
+
     deviceConfig.system.min_loglevel = Level.DEBUG.value();
     checkThatHasInterestingSystemStatus(false);
     untilTrue("last_config updated",
         () -> !dateEquals(stableConfig, deviceState.system.last_config)
     );
     assertTrue("system operational", deviceState.system.operation.operational);
     untilLogged(SYSTEM_CONFIG_APPLY, SYSTEM_CONFIG_APPLY_LEVEL);
+    // These should not be logged since the level was at INFO until the new config is applied.
     checkNotLogged(SYSTEM_CONFIG_RECEIVE, SYSTEM_CONFIG_RECEIVE_LEVEL);
     checkNotLogged(SYSTEM_CONFIG_PARSE, SYSTEM_CONFIG_PARSE_LEVEL);
   }