Do not schedule new context unless already waiting for one

3 files changed, 61 insertions, 20 deletions

diff --git a/node-admin/src/main/java/com/yahoo/vespa/hosted/node/admin/nodeadmin/NodeAdminStateUpdater.java b/node-admin/src/main/java/com/yahoo/vespa/hosted/node/admin/nodeadmin/NodeAdminStateUpdater.java
index 453f4f4b4c1..9d44f125efc 100644
--- a/node-admin/src/main/java/com/yahoo/vespa/hosted/node/admin/nodeadmin/NodeAdminStateUpdater.java
+++ b/node-admin/src/main/java/com/yahoo/vespa/hosted/node/admin/nodeadmin/NodeAdminStateUpdater.java
@@ -27,7 +27,6 @@ import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
 import java.util.function.Supplier;
-import java.util.logging.Level;
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
 
diff --git a/node-admin/src/main/java/com/yahoo/vespa/hosted/node/admin/nodeagent/NodeAgentContextManager.java b/node-admin/src/main/java/com/yahoo/vespa/hosted/node/admin/nodeagent/NodeAgentContextManager.java
index 237b7d0daf7..a86b3865452 100644
--- a/node-admin/src/main/java/com/yahoo/vespa/hosted/node/admin/nodeagent/NodeAgentContextManager.java
+++ b/node-admin/src/main/java/com/yahoo/vespa/hosted/node/admin/nodeagent/NodeAgentContextManager.java
@@ -22,6 +22,7 @@ public class NodeAgentContextManager implements NodeAgentContextSupplier, NodeAg
     private boolean wantFrozen = false;
     private boolean isFrozen = true;
     private boolean pendingInterrupt = false;
+    private boolean waitingForContext = false;
 
     public NodeAgentContextManager(Clock clock, NodeAgentContext context) {
         this.clock = clock;
@@ -31,6 +32,9 @@ public class NodeAgentContextManager implements NodeAgentContextSupplier, NodeAg
     @Override
     public void scheduleTickWith(NodeAgentContext context, Instant at) {
         synchronized (monitor) {
+            // Do not schedule a new context while NodeAgent is still converging
+            if (!waitingForContext) return;
+
             nextContext = Objects.requireNonNull(context);
             nextContextAt = Objects.requireNonNull(at);
             monitor.notifyAll(); // Notify of new context
@@ -61,12 +65,14 @@ public class NodeAgentContextManager implements NodeAgentContextSupplier, NodeAg
     @Override
     public NodeAgentContext nextContext() throws InterruptedException {
         synchronized (monitor) {
+            waitingForContext = true;
             Duration untilNextContext = Duration.ZERO;
             while (setAndGetIsFrozen(wantFrozen) ||
                     nextContext == null ||
                     (untilNextContext = Duration.between(Instant.now(), nextContextAt)).toMillis() > 0) {
                 if (pendingInterrupt) {
                     pendingInterrupt = false;
+                    waitingForContext = false;
                     throw new InterruptedException("interrupt() was called before next context was scheduled");
                 }
 
@@ -77,6 +83,7 @@ public class NodeAgentContextManager implements NodeAgentContextSupplier, NodeAg
 
             currentContext = nextContext;
             nextContext = null;
+            waitingForContext = false;
             return currentContext;
         }
     }
diff --git a/node-admin/src/test/java/com/yahoo/vespa/hosted/node/admin/nodeagent/NodeAgentContextManagerTest.java b/node-admin/src/test/java/com/yahoo/vespa/hosted/node/admin/nodeagent/NodeAgentContextManagerTest.java
index 3771774c9a5..99111366b8e 100644
--- a/node-admin/src/test/java/com/yahoo/vespa/hosted/node/admin/nodeagent/NodeAgentContextManagerTest.java
+++ b/node-admin/src/test/java/com/yahoo/vespa/hosted/node/admin/nodeagent/NodeAgentContextManagerTest.java
@@ -25,23 +25,32 @@ public class NodeAgentContextManagerTest {
     private final NodeAgentContextManager manager = new NodeAgentContextManager(clock, initialContext);
 
     @Test(timeout = TIMEOUT)
-    public void returns_immediately_if_next_context_is_ready() throws InterruptedException {
+    public void context_is_ignored_unless_scheduled_while_waiting() throws InterruptedException {
         NodeAgentContext context1 = generateContext();
         manager.scheduleTickWith(context1, clock.instant());
-
         assertSame(initialContext, manager.currentContext());
-        assertSame(context1, manager.nextContext());
-        assertSame(context1, manager.currentContext());
+
+        AsyncExecutor<NodeAgentContext> async = new AsyncExecutor<>(manager::nextContext);
+        Thread.sleep(20);
+        assertFalse(async.isCompleted());
+
+        NodeAgentContext context2 = generateContext();
+        manager.scheduleTickWith(context2, clock.instant());
+
+        assertSame(context2, async.awaitResult().response.get());
+        assertSame(context2, manager.currentContext());
     }
 
     @Test(timeout = TIMEOUT)
     public void returns_no_earlier_than_at_given_time() throws InterruptedException {
+        AsyncExecutor<NodeAgentContext> async = new AsyncExecutor<>(manager::nextContext);
+        Thread.sleep(20);
+
         NodeAgentContext context1 = generateContext();
         Instant returnAt = clock.instant().plusMillis(500);
         manager.scheduleTickWith(context1, returnAt);
 
-        assertSame(initialContext, manager.currentContext());
-        assertSame(context1, manager.nextContext());
+        assertSame(context1, async.awaitResult().response.get());
         assertSame(context1, manager.currentContext());
         // Is accurate to a millisecond
         assertFalse(clock.instant().plusMillis(1).isBefore(returnAt));
@@ -50,9 +59,9 @@ public class NodeAgentContextManagerTest {
     @Test(timeout = TIMEOUT)
     public void blocks_in_nextContext_until_one_is_scheduled() throws InterruptedException {
         AsyncExecutor<NodeAgentContext> async = new AsyncExecutor<>(manager::nextContext);
-        assertFalse(async.response.isPresent());
+        assertFalse(async.isCompleted());
         Thread.sleep(10);
-        assertFalse(async.response.isPresent());
+        assertFalse(async.isCompleted());
 
         NodeAgentContext context1 = generateContext();
         manager.scheduleTickWith(context1, clock.instant());
@@ -65,9 +74,9 @@ public class NodeAgentContextManagerTest {
     @Test(timeout = TIMEOUT)
     public void blocks_in_nextContext_until_interrupt() throws InterruptedException {
         AsyncExecutor<NodeAgentContext> async = new AsyncExecutor<>(manager::nextContext);
-        assertFalse(async.response.isPresent());
+        assertFalse(async.isCompleted());
         Thread.sleep(10);
-        assertFalse(async.response.isPresent());
+        assertFalse(async.isCompleted());
 
         manager.interrupt();
 
@@ -82,8 +91,10 @@ public class NodeAgentContextManagerTest {
 
         // Generate new context and get it from the supplier, this completes the unfreeze
         NodeAgentContext context1 = generateContext();
+        AsyncExecutor<NodeAgentContext> async = new AsyncExecutor<>(manager::nextContext);
+        Thread.sleep(20);
         manager.scheduleTickWith(context1, clock.instant());
-        assertSame(context1, manager.nextContext());
+        assertSame(context1, async.awaitResult().response.get());
 
         assertTrue(manager.setFrozen(false, Duration.ZERO));
     }
@@ -100,17 +111,34 @@ public class NodeAgentContextManagerTest {
 
     @Test(timeout = TIMEOUT)
     public void setFrozen_is_successful_if_converged_in_time() throws InterruptedException {
-        AsyncExecutor<Boolean> async = new AsyncExecutor<>(() -> manager.setFrozen(false, Duration.ofMillis(500)));
-
-        assertFalse(async.response.isPresent());
+        AsyncExecutor<NodeAgentContext> asyncConsumer1 = new AsyncExecutor<>(() -> {
+            NodeAgentContext context = manager.nextContext();
+            Thread.sleep(200); // Simulate running NodeAgent::converge
+            return context;
+        });
+        Thread.sleep(20);
 
         NodeAgentContext context1 = generateContext();
         manager.scheduleTickWith(context1, clock.instant());
-        assertSame(context1, manager.nextContext());
+        Thread.sleep(10);
 
-        async.awaitResult();
-        assertEquals(Optional.of(true), async.response);
-        assertFalse(async.exception.isPresent());
+        // Scheduler wants to freeze
+        AsyncExecutor<Boolean> asyncScheduler = new AsyncExecutor<>(() -> manager.setFrozen(true, Duration.ofMillis(500)));
+        Thread.sleep(20);
+        assertFalse(asyncConsumer1.isCompleted()); // Still running NodeAgent::converge
+        assertSame(context1, asyncConsumer1.awaitResult().response.get());
+        assertFalse(asyncScheduler.isCompleted()); // Still waiting for consumer to converge to frozen
+
+        AsyncExecutor<NodeAgentContext> asyncConsumer2 = new AsyncExecutor<>(manager::nextContext);
+        Thread.sleep(20);
+        assertFalse(asyncConsumer2.isCompleted()); // Waiting for next context
+        assertTrue(asyncScheduler.isCompleted()); // While consumer is waiting, it has converged to frozen
+
+        // Interrupt manager to end asyncConsumer2
+        manager.interrupt();
+        asyncConsumer2.awaitResult();
+
+        assertEquals(Optional.of(true), asyncScheduler.response);
     }
 
     private static NodeAgentContext generateContext() {
@@ -139,7 +167,7 @@ public class NodeAgentContextManagerTest {
             this.thread.start();
         }
 
-        private void awaitResult() {
+        private AsyncExecutor<T> awaitResult() {
             synchronized (monitor) {
                 while (!completed) {
                     try {
@@ -147,6 +175,13 @@ public class NodeAgentContextManagerTest {
                     } catch (InterruptedException ignored) { }
                 }
             }
+            return this;
+        }
+
+        private boolean isCompleted() {
+            synchronized (monitor) {
+                return completed;
+            }
         }
     }