Changing Workflow partitioning causes trigger time offsets
When changing a Workflows partitioning configuration, the previously stored `nextNaturalTrigger` time is kept, which potentially leads to an offset trigger time.

Reproducing
- found an hourly Workflow that was enabled
- nextNaturalTrigger: 2016-12-08 (19:00:00.000) UTC
- changed to daily
- it was triggered at 2016-12-08 (19:00:00.000) UTC
- nextNaturalTrigger: 2016-12-09 (19:00:00.000) UTC

The workflow is now offset to 19:00 UTC instead of 00:00 which a daily workflow would trigger at normally.
diff --git a/styx-api-service/src/test/java/com/spotify/styx/api/WorkflowResourceTest.java b/styx-api-service/src/test/java/com/spotify/styx/api/WorkflowResourceTest.java
index 7d47ebbe..c11a0ded 100644
--- a/styx-api-service/src/test/java/com/spotify/styx/api/WorkflowResourceTest.java
+++ b/styx-api-service/src/test/java/com/spotify/styx/api/WorkflowResourceTest.java
@@ -270,7 +270,9 @@ public class WorkflowResourceTest {
     assertNoJson(response, "docker_image");
     assertNoJson(response, "commit_sha");
 
-    storage.patchState(WORKFLOW.id(), WorkflowState.all(true, "tina:ranic", "470a229b49a14e7682af2abfdac3b881a8aacdf9"));
+    storage.patchState(WORKFLOW.id(),
+        WorkflowState.builder().enabled(true).dockerImage("tina:ranic")
+            .commitSha("470a229b49a14e7682af2abfdac3b881a8aacdf9").build());
 
     response =
         awaitResponse(serviceHelper.request("GET", path("/foo/bar/state")));
diff --git a/styx-common/src/main/java/com/spotify/styx/model/WorkflowState.java b/styx-common/src/main/java/com/spotify/styx/model/WorkflowState.java
index 6f7bbe46..7af85f78 100644
--- a/styx-common/src/main/java/com/spotify/styx/model/WorkflowState.java
+++ b/styx-common/src/main/java/com/spotify/styx/model/WorkflowState.java
@@ -20,12 +20,11 @@
 
 package com.spotify.styx.model;
 
-import static java.util.Optional.of;
-
 import com.fasterxml.jackson.annotation.JsonCreator;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.auto.value.AutoValue;
+import java.time.Instant;
 import java.util.Optional;
 
 /**
@@ -44,27 +43,53 @@ public abstract class WorkflowState {
   @JsonProperty
   public abstract Optional<String> commitSha();
 
+  @JsonProperty
+  public abstract Optional<Instant> nextNaturalTrigger();
+
+  public abstract Builder toBuilder();
+
+  public static Builder builder() {
+    return new AutoValue_WorkflowState.Builder();
+  }
+
+  @AutoValue.Builder
+  public abstract static class Builder {
+
+    public abstract Builder enabled(boolean enabled);
+    public abstract Builder dockerImage(String dockerImage);
+    public abstract Builder commitSha(String commitSha);
+    public abstract Builder nextNaturalTrigger(Instant nextNaturalTrigger);
+
+    public abstract WorkflowState build();
+  }
+
   @JsonCreator
   public static WorkflowState create(
       @JsonProperty("enabled") Optional<Boolean> enabled,
       @JsonProperty("docker_image") Optional<String> dockerImage,
-      @JsonProperty("commit_sha") Optional<String> commitSha) {
-    return new AutoValue_WorkflowState(enabled, dockerImage, commitSha);
+      @JsonProperty("commit_sha") Optional<String> commitSha,
+      @JsonProperty("next_natural_trigger") Optional<Instant> nextNaturalTrigger) {
+    Builder builder = builder();
+    enabled.ifPresent(builder::enabled);
+    dockerImage.ifPresent(builder::dockerImage);
+    commitSha.ifPresent(builder::commitSha);
+    nextNaturalTrigger.ifPresent(builder::nextNaturalTrigger);
+    return builder.build();
   }
 
   public static WorkflowState all(boolean enabled, String dockerImage, String commitSha) {
-    return create(of(enabled), of(dockerImage), of(commitSha));
+    return builder().enabled(enabled).dockerImage(dockerImage).commitSha(commitSha).build();
   }
 
   public static WorkflowState empty() {
-    return create(Optional.empty(), Optional.empty(), Optional.empty());
+    return builder().build();
   }
 
   public static WorkflowState patchDockerImage(String dockerImage) {
-    return create(Optional.empty(), of(dockerImage), Optional.empty());
+    return builder().dockerImage(dockerImage).build();
   }
 
   public static WorkflowState patchEnabled(boolean enabled) {
-    return create(of(enabled), Optional.empty(), Optional.empty());
+    return builder().enabled(enabled).build();
   }
 }
diff --git a/styx-common/src/main/java/com/spotify/styx/storage/DatastoreStorage.java b/styx-common/src/main/java/com/spotify/styx/storage/DatastoreStorage.java
index 8cffa1fc..f374493b 100644
--- a/styx-common/src/main/java/com/spotify/styx/storage/DatastoreStorage.java
+++ b/styx-common/src/main/java/com/spotify/styx/storage/DatastoreStorage.java
@@ -75,7 +75,7 @@ class DatastoreStorage {
   public static final String PROPERTY_CONFIG_DOCKER_RUNNER_ID = "dockerRunnerId";
   public static final String PROPERTY_WORKFLOW_JSON = "json";
   public static final String PROPERTY_WORKFLOW_ENABLED = "enabled";
-  public static final String PROPERTY_NEXT_EXECUTION = "nextNaturalTrigger";
+  public static final String PROPERTY_NEXT_NATURAL_TRIGGER = "nextNaturalTrigger";
   public static final String PROPERTY_DOCKER_IMAGE = "dockerImage";
   public static final String PROPERTY_COUNTER = "counter";
   public static final String PROPERTY_COMPONENT = "component";
@@ -227,7 +227,7 @@ class DatastoreStorage {
 
       final Entity.Builder builder = Entity
           .builder(workflowOpt.get())
-          .set(PROPERTY_NEXT_EXECUTION, instantToDatetime(nextNaturalTrigger));
+          .set(PROPERTY_NEXT_NATURAL_TRIGGER, instantToDatetime(nextNaturalTrigger));
       return transaction.put(builder.build());
     }));
   }
@@ -250,8 +250,8 @@ class DatastoreStorage {
         continue;
       }
       map.put(workflow,
-          entity.contains(PROPERTY_NEXT_EXECUTION)
-          ? Optional.of(datetimeToInstant(entity.getDateTime(PROPERTY_NEXT_EXECUTION)))
+          entity.contains(PROPERTY_NEXT_NATURAL_TRIGGER)
+          ? Optional.of(datetimeToInstant(entity.getDateTime(PROPERTY_NEXT_NATURAL_TRIGGER)))
           : Optional.empty());
     }
     return map;
@@ -321,7 +321,8 @@ class DatastoreStorage {
       state.enabled().ifPresent(x -> builder.set(PROPERTY_WORKFLOW_ENABLED, x));
       state.dockerImage().ifPresent(x -> builder.set(PROPERTY_DOCKER_IMAGE, x));
       state.commitSha().ifPresent(x -> builder.set(PROPERTY_COMMIT_SHA, x));
-
+      state.nextNaturalTrigger()
+          .ifPresent(x -> builder.set(PROPERTY_NEXT_NATURAL_TRIGGER, instantToDatetime(x)));
       return transaction.put(builder.build());
     }));
   }
@@ -359,11 +360,10 @@ class DatastoreStorage {
   }
 
   public WorkflowState workflowState(WorkflowId workflowId) throws IOException {
-    return
-        WorkflowState.create(
-            Optional.of(enabled(workflowId)),
-            getDockerImage(workflowId),
-            getCommitSha(workflowId));
+    WorkflowState.Builder builder = WorkflowState.builder().enabled(enabled(workflowId));
+    getDockerImage(workflowId).ifPresent(builder::dockerImage);
+    getCommitSha(workflowId).ifPresent(builder::commitSha);
+    return builder.build();
   }
 
   private Optional<String> getCommitSha(WorkflowId workflowId) {
diff --git a/styx-common/src/main/java/com/spotify/styx/storage/InMemStorage.java b/styx-common/src/main/java/com/spotify/styx/storage/InMemStorage.java
index 759c04c8..e92e7097 100644
--- a/styx-common/src/main/java/com/spotify/styx/storage/InMemStorage.java
+++ b/styx-common/src/main/java/com/spotify/styx/storage/InMemStorage.java
@@ -187,7 +187,7 @@ public class InMemStorage implements Storage, EventStorage {
     return
         workflowStatePerWorkflowId.getOrDefault(
             workflowId,
-            WorkflowState.create(Optional.of(false), Optional.empty(), Optional.empty()));
+            WorkflowState.patchEnabled(false));
   }
 
   @Override
diff --git a/styx-common/src/main/java/com/spotify/styx/util/WorkflowStateUtil.java b/styx-common/src/main/java/com/spotify/styx/util/WorkflowStateUtil.java
index 16db26df..e60af87e 100644
--- a/styx-common/src/main/java/com/spotify/styx/util/WorkflowStateUtil.java
+++ b/styx-common/src/main/java/com/spotify/styx/util/WorkflowStateUtil.java
@@ -30,11 +30,17 @@ public final class WorkflowStateUtil {
   public static WorkflowState patchWorkflowState(
       final Optional<WorkflowState> originalWorkflowState,
       final WorkflowState patch) {
+
     return originalWorkflowState.map(
-        o -> WorkflowState.create(
-            Optional.of(patch.enabled().orElse(o.enabled().orElse(false))),
-            patch.dockerImage().isPresent() ? patch.dockerImage() : o.dockerImage(),
-            patch.commitSha().isPresent() ? patch.commitSha() : o.commitSha())
+        o -> {
+          WorkflowState.Builder builder = o.toBuilder()
+              .enabled(patch.enabled().orElse(o.enabled().orElse(false)));
+          patch.dockerImage().ifPresent(dockerImage -> builder.dockerImage(dockerImage));
+          patch.commitSha().ifPresent(commitSha -> builder.commitSha(commitSha));
+          patch.nextNaturalTrigger()
+              .ifPresent(nextNaturalTrigger -> builder.nextNaturalTrigger(nextNaturalTrigger));
+          return builder.build();
+        }
     ).orElse(patch);
   }
 }
diff --git a/styx-common/src/test/java/com/spotify/styx/storage/DatastoreStorageTest.java b/styx-common/src/test/java/com/spotify/styx/storage/DatastoreStorageTest.java
index 2b3a9c73..12ad1ac4 100644
--- a/styx-common/src/test/java/com/spotify/styx/storage/DatastoreStorageTest.java
+++ b/styx-common/src/test/java/com/spotify/styx/storage/DatastoreStorageTest.java
@@ -211,7 +211,7 @@ public class DatastoreStorageTest {
 
   @Test
   public void shouldPersistCommitShaPerComponent() throws Exception {
-    WorkflowState state = WorkflowState.create(empty(), empty(), of(COMMIT_SHA));
+    WorkflowState state = WorkflowState.builder().commitSha(COMMIT_SHA).build();
 
     storage.store(Workflow.create(WORKFLOW_ID1.componentId(), URI.create("http://foo"), DATA_ENDPOINT_EMPTY_CONF));
     storage.patchState(WORKFLOW_ID1.componentId(), state);
@@ -236,7 +236,7 @@ public class DatastoreStorageTest {
     storage.store(Workflow.create(WORKFLOW_ID1.componentId(), URI.create("http://foo"), DataEndpoint
         .create(WORKFLOW_ID1.endpointId(), Partitioning.DAYS, Optional.empty(), Optional.empty(),
                 Optional.empty(), emptyList())));
-    storage.patchState(WORKFLOW_ID1, WorkflowState.create(empty(), empty(), of(COMMIT_SHA)));
+    storage.patchState(WORKFLOW_ID1, WorkflowState.builder().commitSha(COMMIT_SHA).build());
     WorkflowState retrieved = storage.workflowState(WORKFLOW_ID1);
 
     assertThat(retrieved.commitSha(), is(Optional.of(COMMIT_SHA)));
@@ -254,7 +254,7 @@ public class DatastoreStorageTest {
   public void shouldReturnEmptyWorkflowStateExceptEnabledWhenWorkflowStateDoesNotExist() throws Exception {
     storage.store(WORKFLOW_NO_STATE);
     WorkflowState retrieved = storage.workflowState(WORKFLOW_ID_NO_STATE);
-    assertThat(retrieved, is(WorkflowState.create(Optional.of(false),Optional.empty(), Optional.empty())));
+    assertThat(retrieved, is(WorkflowState.patchEnabled(false)));
   }
 
   @Test
@@ -486,7 +486,11 @@ public class DatastoreStorageTest {
     storage.store(Workflow.create(WORKFLOW_ID1.componentId(), URI.create("http://not/important"), DataEndpoint
         .create(WORKFLOW_ID1.endpointId(), Partitioning.DAYS, Optional.empty(), Optional.empty(),
                 Optional.empty(), emptyList())));
-    WorkflowState state = WorkflowState.all(true, DOCKER_IMAGE.get() , COMMIT_SHA);
+    WorkflowState state = WorkflowState.builder()
+        .enabled(true)
+        .dockerImage(DOCKER_IMAGE.get())
+        .commitSha(COMMIT_SHA)
+        .build();
     storage.patchState(WORKFLOW_ID1, state);
 
     WorkflowState retrieved = storage.workflowState(WORKFLOW_ID1);
diff --git a/styx-common/src/test/java/com/spotify/styx/util/WorkflowStateUtilTest.java b/styx-common/src/test/java/com/spotify/styx/util/WorkflowStateUtilTest.java
index b2978915..54c9e8b7 100644
--- a/styx-common/src/test/java/com/spotify/styx/util/WorkflowStateUtilTest.java
+++ b/styx-common/src/test/java/com/spotify/styx/util/WorkflowStateUtilTest.java
@@ -50,7 +50,7 @@ public class WorkflowStateUtilTest {
 
   @Test
   public void patchEnabledFieldReturnsPatchedOriginal() {
-    WorkflowState patch = WorkflowState.create(Optional.of(false), Optional.empty(), Optional.empty());
+    WorkflowState patch = WorkflowState.builder().enabled(false).build();
     WorkflowState patchedState = WorkflowStateUtil.patchWorkflowState(
         Optional.of(FULLY_POPULATED_STATE),
         patch);
@@ -62,17 +62,17 @@ public class WorkflowStateUtilTest {
     WorkflowState patch = WorkflowState.empty();
     WorkflowState patchedState = WorkflowStateUtil.patchWorkflowState(
         Optional.of(
-            WorkflowState.create(
-                Optional.empty(),
-                Optional.of("original_docker_image"),
-                Optional.of(ORIGINAL_COMMIT_SHA))),
+            WorkflowState.builder()
+                .dockerImage("original_docker_image")
+                .commitSha(ORIGINAL_COMMIT_SHA)
+                .build()),
         patch);
     assertThat(patchedState, equalTo(WorkflowState.all(false, "original_docker_image", ORIGINAL_COMMIT_SHA)));
   }
 
   @Test
   public void patchDockerImageFieldReturnsPatchedOriginal() {
-    WorkflowState patch = WorkflowState.create(Optional.empty(), Optional.of("patched_docker_image"), Optional.empty());
+    WorkflowState patch = WorkflowState.builder().dockerImage("patched_docker_image").build();
     WorkflowState patchedState = WorkflowStateUtil.patchWorkflowState(
         Optional.of(FULLY_POPULATED_STATE),
         patch);
@@ -81,11 +81,13 @@ public class WorkflowStateUtilTest {
 
   @Test
   public void patchCommitShaFieldReturnsPatchedOriginal() {
-    WorkflowState patch = WorkflowState.create(Optional.empty(), Optional.empty(), Optional.of(PATCHED_COMMIT_SHA));
+    WorkflowState patch = WorkflowState.builder().commitSha(PATCHED_COMMIT_SHA).build();
     WorkflowState patchedState = WorkflowStateUtil.patchWorkflowState(
         Optional.of(FULLY_POPULATED_STATE),
         patch);
-    assertThat(patchedState, equalTo(WorkflowState.all(true, "original_docker_image", PATCHED_COMMIT_SHA)));
+    assertThat(patchedState, equalTo(
+        WorkflowState.builder().enabled(true).dockerImage("original_docker_image")
+            .commitSha(PATCHED_COMMIT_SHA).build()));
   }
 
 }
diff --git a/styx-scheduler-service/src/main/java/com/spotify/styx/StyxScheduler.java b/styx-scheduler-service/src/main/java/com/spotify/styx/StyxScheduler.java
index b906a84c..dab9bee1 100644
--- a/styx-scheduler-service/src/main/java/com/spotify/styx/StyxScheduler.java
+++ b/styx-scheduler-service/src/main/java/com/spotify/styx/StyxScheduler.java
@@ -24,6 +24,8 @@ import static com.spotify.styx.util.Connections.createBigTableConnection;
 import static com.spotify.styx.util.Connections.createDatastore;
 import static com.spotify.styx.util.ReplayEvents.replayActiveStates;
 import static com.spotify.styx.util.ReplayEvents.transitionLogger;
+import static com.spotify.styx.workflow.ParameterUtil.incrementInstant;
+import static com.spotify.styx.workflow.ParameterUtil.truncateInstant;
 import static java.util.Objects.requireNonNull;
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static java.util.stream.Collectors.toMap;
@@ -50,9 +52,11 @@ import com.spotify.styx.api.SchedulerResource;
 import com.spotify.styx.docker.DockerRunner;
 import com.spotify.styx.docker.WorkflowValidator;
 import com.spotify.styx.model.Event;
+import com.spotify.styx.model.Partitioning;
 import com.spotify.styx.model.Workflow;
 import com.spotify.styx.model.WorkflowId;
 import com.spotify.styx.model.WorkflowInstance;
+import com.spotify.styx.model.WorkflowState;
 import com.spotify.styx.monitoring.MeteredDockerRunner;
 import com.spotify.styx.monitoring.MeteredEventStorage;
 import com.spotify.styx.monitoring.MeteredStorage;
@@ -92,6 +96,7 @@ import java.time.Duration;
 import java.time.Instant;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.ServiceLoader;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
@@ -323,7 +328,7 @@ public class StyxScheduler implements AppInit {
 
     final WorkflowCache cache = new InMemWorkflowCache();
     final Consumer<Workflow> workflowChangeListener = workflowChanged(cache, storage,
-                                                                      stats, stateManager);
+                                                                      stats, stateManager, time);
     final Consumer<Workflow> workflowRemoveListener = workflowRemoved(storage);
 
     restoreState(eventStorage, outputHandlers, stateManager);
@@ -485,7 +490,8 @@ public class StyxScheduler implements AppInit {
       WorkflowCache cache,
       Storage storage,
       Stats stats,
-      StateManager stateManager) {
+      StateManager stateManager,
+      Time time) {
 
     return (workflow) -> {
       stats.registerActiveStates(
@@ -494,7 +500,21 @@ public class StyxScheduler implements AppInit {
 
       cache.store(workflow);
       try {
+        Optional<Workflow> optWorkflow = storage.workflow(workflow.id());
         storage.store(workflow);
+
+        // update nextNaturalTrigger only when partitioning specification changes.
+        final Partitioning partitioning = workflow.schedule().partitioning();
+        if (optWorkflow.isPresent() && !optWorkflow.get().schedule().partitioning()
+            .equals(partitioning)) {
+          final Instant nextNaturalTrigger =
+              incrementInstant(truncateInstant(time.get(), partitioning),
+                  partitioning);
+          storage.patchState(workflow.id(),
+              WorkflowState.builder()
+                  .nextNaturalTrigger(nextNaturalTrigger)
+                  .build());
+        }
       } catch (IOException e) {
         LOG.warn("Failed to store workflow " + workflow, e);
       }
diff --git a/styx-scheduler-service/src/test/java/com/spotify/styx/StyxSchedulerServiceFixture.java b/styx-scheduler-service/src/test/java/com/spotify/styx/StyxSchedulerServiceFixture.java
index 39d85ec0..c791f400 100644
--- a/styx-scheduler-service/src/test/java/com/spotify/styx/StyxSchedulerServiceFixture.java
+++ b/styx-scheduler-service/src/test/java/com/spotify/styx/StyxSchedulerServiceFixture.java
@@ -187,6 +187,18 @@ public class StyxSchedulerServiceFixture {
     workflowRemoveListener.accept(workflow);
   }
 
+  /**
+   * Fast forwards the time without any execution in-between.
+   */
+  void timeJumps(int n, TimeUnit unit) {
+    LOG.info("{} {} passes", n, unit);
+    now = now.plusMillis(unit.toMillis(n));
+    printTime();
+  }
+
+  /**
+   * Fast forwards the time by executing all tasks in-between according to the executor's delay.
+   */
   void timePasses(int n, TimeUnit unit) {
     LOG.info("{} {} passes", n, unit);
     now = now.plusMillis(unit.toMillis(n));
@@ -247,6 +259,12 @@ public class StyxSchedulerServiceFixture {
     await().until(() -> dockerRuns.size() == n);
   }
 
+  void awaitNumberOfDockerRunsWontChange(int n) {
+    await().pollDelay(1, TimeUnit.SECONDS).until(() -> {
+      return dockerRuns.size() == n;
+    });
+  }
+
   void awaitWorkflowInstanceState(WorkflowInstance instance, RunState.State state) {
     await().until(() -> {
       final RunState runState = getState(instance);
@@ -254,6 +272,10 @@ public class StyxSchedulerServiceFixture {
     });
   }
 
+  void awaitWorkflowInstanceCompletion(WorkflowInstance workflowInstance) {
+    await().until(() -> getState(workflowInstance) == null);
+  }
+
   private void printTime() {
     LOG.info("The time is {}", now);
   }
diff --git a/styx-scheduler-service/src/test/java/com/spotify/styx/SystemTest.java b/styx-scheduler-service/src/test/java/com/spotify/styx/SystemTest.java
index 50106d91..6495cddb 100644
--- a/styx-scheduler-service/src/test/java/com/spotify/styx/SystemTest.java
+++ b/styx-scheduler-service/src/test/java/com/spotify/styx/SystemTest.java
@@ -25,6 +25,8 @@ import static java.util.Arrays.asList;
 import static java.util.Collections.emptyList;
 import static java.util.Optional.empty;
 import static java.util.Optional.of;
+import static java.util.concurrent.TimeUnit.DAYS;
+import static java.util.concurrent.TimeUnit.HOURS;
 import static java.util.concurrent.TimeUnit.MINUTES;
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.hamcrest.Matchers.contains;
@@ -44,22 +46,30 @@ import com.spotify.styx.state.handlers.TerminationHandler;
 import com.spotify.styx.testdata.TestData;
 import java.time.Instant;
 import java.util.Arrays;
+import java.util.Optional;
 import org.junit.Test;
 
 public class SystemTest extends StyxSchedulerServiceFixture {
 
-  private static final DataEndpoint DATA_ENDPOINT = DataEndpoint.create(
+  private static final DataEndpoint DATA_ENDPOINT_HOURLY = DataEndpoint.create(
       "styx.TestEndpoint", Partitioning.HOURS, of("busybox"), of(asList("--hour", "{}")),
       empty(), emptyList());
+  private static final DataEndpoint DATA_ENDPOINT_DAILY = DataEndpoint.create(
+      "styx.TestEndpoint", Partitioning.DAYS, of("busybox"), of(asList("--hour", "{}")),
+      empty(), emptyList());
   private static final String TEST_EXECUTION_ID_1 = "execution_1";
   private static final String TEST_DOCKER_IMAGE = "busybox:1.1";
   private static final Workflow HOURLY_WORKFLOW = Workflow.create(
       "styx",
       TestData.WORKFLOW_URI,
-      DATA_ENDPOINT);
+      DATA_ENDPOINT_HOURLY);
   private static final ExecutionDescription TEST_EXECUTION_DESCRIPTION =
       ExecutionDescription.create(
           TEST_DOCKER_IMAGE, Arrays.asList("--date", "{}", "--bar"), empty(), empty());
+  private static final Workflow DAILY_WORKFLOW = Workflow.create(
+      "styx",
+      TestData.WORKFLOW_URI,
+      DATA_ENDPOINT_DAILY);
 
   @Test
   public void shouldCatchUpWithNaturalTriggers() throws Exception {
@@ -132,6 +142,110 @@ public class SystemTest extends StyxSchedulerServiceFixture {
     assertThat(getState(instance2), is(nullValue()));
   }
 
+  @Test
+  public void updatesNextNaturalTriggerWhenWFPartitioningChangesFromFinerToCoarser() throws Exception {
+    givenTheTimeIs("2016-03-14T15:30:00Z");
+    givenTheGlobalEnableFlagIs(true);
+    givenWorkflow(HOURLY_WORKFLOW);
+    givenWorkflowEnabledStateIs(HOURLY_WORKFLOW, true);
+    WorkflowInstance workflowInstance =
+        WorkflowInstance.create(HOURLY_WORKFLOW.id(), "2016-03-14T14");
+
+    styxStarts();
+    timePasses(1, SECONDS);
+    awaitWorkflowInstanceState(workflowInstance, RunState.State.QUEUED);
+    timePasses(1, SECONDS);
+    awaitNumberOfDockerRuns(1);
+
+    workflowInstance = dockerRuns.get(0)._1;
+    RunSpec runSpec = dockerRuns.get(0)._2;
+    assertThat(workflowInstance.workflowId(), is(HOURLY_WORKFLOW.id()));
+    assertThat(runSpec, is(RunSpec.simple("busybox", "--hour", "2016-03-14T14")));
+
+    workflowInstance = WorkflowInstance.create(HOURLY_WORKFLOW.id(), "2016-03-14");
+    // this should store a new value for nextNaturalTrigger, 2016-03-15.
+    workflowChanges(DAILY_WORKFLOW);
+    timeJumps(1, DAYS);
+    timePasses(1, SECONDS);
+    awaitWorkflowInstanceState(workflowInstance, RunState.State.QUEUED);
+    timePasses(1, SECONDS);
+    awaitNumberOfDockerRuns(2);
+
+    workflowInstance = dockerRuns.get(1)._1;
+    runSpec = dockerRuns.get(1)._2;
+    assertThat(workflowInstance.workflowId(), is(DAILY_WORKFLOW.id()));
+    assertThat(runSpec, is(RunSpec.simple("busybox", "--hour", "2016-03-14")));
+  }
+
+  @Test
+  public void updatesNextNaturalTriggerWhenWFPartitioningChangesFromCoarserToFiner() throws Exception {
+    givenTheTimeIs("2016-03-14T15:30:00Z");
+    givenTheGlobalEnableFlagIs(true);
+    givenWorkflow(DAILY_WORKFLOW);
+    givenWorkflowEnabledStateIs(DAILY_WORKFLOW, true);
+    WorkflowInstance workflowInstance = WorkflowInstance.create(DAILY_WORKFLOW.id(), "2016-03-13");
+
+    styxStarts();
+    timePasses(1, SECONDS);
+    awaitWorkflowInstanceState(workflowInstance, RunState.State.QUEUED);
+    timePasses(1, SECONDS);
+    awaitNumberOfDockerRuns(1);
+
+    workflowInstance = dockerRuns.get(0)._1;
+    RunSpec runSpec = dockerRuns.get(0)._2;
+    assertThat(workflowInstance.workflowId(), is(DAILY_WORKFLOW.id()));
+    assertThat(runSpec, is(RunSpec.simple("busybox", "--hour", "2016-03-13")));
+
+    workflowInstance = WorkflowInstance.create(HOURLY_WORKFLOW.id(), "2016-03-14T15");
+    // this should store a new value for nextNaturalTrigger, 2016-03-14T16.
+    workflowChanges(HOURLY_WORKFLOW);
+    timeJumps(1, HOURS);
+    timePasses(1, SECONDS);
+    awaitWorkflowInstanceState(workflowInstance, RunState.State.QUEUED);
+    timePasses(1, SECONDS);
+
+    awaitNumberOfDockerRuns(2);
+    workflowInstance = dockerRuns.get(1)._1;
+    runSpec = dockerRuns.get(1)._2;
+    assertThat(workflowInstance.workflowId(), is(DAILY_WORKFLOW.id()));
+    assertThat(runSpec, is(RunSpec.simple("busybox", "--hour", "2016-03-14T15")));
+  }
+
+  @Test
+  public void doesntUpdateNextNaturalTriggerWhenPartitioningDoesntChange() throws Exception {
+    givenTheTimeIs("2016-03-14T15:30:00Z");
+    givenTheGlobalEnableFlagIs(true);
+    givenWorkflow(DAILY_WORKFLOW);
+    givenWorkflowEnabledStateIs(DAILY_WORKFLOW, true);
+    WorkflowInstance workflowInstance =
+        WorkflowInstance.create(DAILY_WORKFLOW.id(), "2016-03-13");
+
+    styxStarts();
+    timePasses(1, SECONDS);
+    awaitWorkflowInstanceState(workflowInstance, RunState.State.QUEUED);
+    timePasses(1, SECONDS);
+    awaitNumberOfDockerRuns(1);
+
+    workflowInstance = dockerRuns.get(0)._1;
+    RunSpec runSpec = dockerRuns.get(0)._2;
+    assertThat(workflowInstance.workflowId(), is(DAILY_WORKFLOW.id()));
+    assertThat(runSpec, is(RunSpec.simple("busybox", "--hour", "2016-03-13")));
+
+    injectEvent(Event.started(workflowInstance));
+    injectEvent(Event.terminate(workflowInstance, 0));
+    awaitWorkflowInstanceCompletion(workflowInstance);
+
+    workflowChanges(Workflow.create(DAILY_WORKFLOW.componentId(),
+        DAILY_WORKFLOW.componentUri(),
+        DataEndpoint.create(DATA_ENDPOINT_DAILY.id(), DATA_ENDPOINT_DAILY.partitioning(),
+            Optional.of("freebox"), DATA_ENDPOINT_DAILY.dockerArgs(),
+            DATA_ENDPOINT_DAILY.secret(), emptyList())));
+    timePasses(StyxScheduler.SCHEDULER_TICK_INTERVAL_SECONDS, SECONDS);
+    awaitNumberOfDockerRunsWontChange(1);
+
+    assertThat(dockerRuns.size(), is(1));
+  }
+
   @Test
   public void runsDockerImageWithArgsTemplate() throws Exception {
     givenTheTimeIs("2016-03-14T15:59:00Z");
@@ -172,8 +286,8 @@ public class SystemTest extends StyxSchedulerServiceFixture {
     awaitWorkflowInstanceState(workflowInstance, RunState.State.QUEUED);
 
     DataEndpoint changedDataEndpoint = DataEndpoint.create(
-        DATA_ENDPOINT.id(), Partitioning.HOURS, of("busybox:v777"), of(asList("other", "args")),
-        empty(), emptyList());
+        DATA_ENDPOINT_HOURLY.id(), Partitioning.HOURS, of("busybox:v777"),
+        of(asList("other", "args")), empty(), emptyList());
 
     Workflow changedWorkflow = Workflow.create(
         HOURLY_WORKFLOW.componentId(),
diff --git a/styx-scheduler-service/src/test/java/com/spotify/styx/state/handlers/ExecutionDescriptionHandlerTest.java b/styx-scheduler-service/src/test/java/com/spotify/styx/state/handlers/ExecutionDescriptionHandlerTest.java
index 0348cdce..632e4dbe 100644
--- a/styx-scheduler-service/src/test/java/com/spotify/styx/state/handlers/ExecutionDescriptionHandlerTest.java
+++ b/styx-scheduler-service/src/test/java/com/spotify/styx/state/handlers/ExecutionDescriptionHandlerTest.java
@@ -70,8 +70,11 @@ public class ExecutionDescriptionHandlerTest {
   @Test
   public void shouldTransitionIntoSubmitting() throws Exception {
     Workflow workflow = Workflow.create("id", TestData.WORKFLOW_URI, dataEndpoint("--date", "{}", "--bar"));
-    WorkflowState workflowState = WorkflowState.create(
-        Optional.of(true), Optional.of(DOCKER_IMAGE), Optional.of(COMMIT_SHA));
+    WorkflowState workflowState = WorkflowState.builder()
+        .enabled(true)
+        .dockerImage(DOCKER_IMAGE)
+        .commitSha(COMMIT_SHA)
+        .build();
     WorkflowInstance workflowInstance = WorkflowInstance.create(workflow.id(), "2016-03-14");
     RunState runState = RunState.fresh(workflowInstance, toTest);
 
@@ -94,8 +97,11 @@ public class ExecutionDescriptionHandlerTest {
   @Test
   public void shouldTransitionIntoFailedIfStorageError() throws Exception {
     Workflow workflow = Workflow.create("id", TestData.WORKFLOW_URI, dataEndpoint("--date", "{}", "--bar"));
-    WorkflowState workflowState = WorkflowState.create(
-        Optional.of(true), Optional.of(DOCKER_IMAGE), Optional.of(COMMIT_SHA));
+    WorkflowState workflowState = WorkflowState.builder()
+        .enabled(true)
+        .dockerImage(DOCKER_IMAGE)
+        .commitSha(COMMIT_SHA)
+        .build();
     WorkflowInstance workflowInstance = WorkflowInstance.create(workflow.id(), "2016-03-14");
 
     Storage storageSpy = Mockito.spy(storage);
['styx-api-service/src/test/java/com/spotify/styx/api/WorkflowResourceTest.java', 'styx-scheduler-service/src/test/java/com/spotify/styx/StyxSchedulerServiceFixture.java', 'styx-common/src/main/java/com/spotify/styx/storage/DatastoreStorage.java', 'styx-common/src/main/java/com/spotify/styx/util/WorkflowStateUtil.java', 'styx-common/src/main/java/com/spotify/styx/storage/InMemStorage.java', 'styx-scheduler-service/src/main/java/com/spotify/styx/StyxScheduler.java', 'styx-scheduler-service/src/test/java/com/spotify/styx/SystemTest.java', 'styx-common/src/test/java/com/spotify/styx/util/WorkflowStateUtilTest.java', 'styx-common/src/main/java/com/spotify/styx/model/WorkflowState.java', 'styx-scheduler-service/src/test/java/com/spotify/styx/state/handlers/ExecutionDescriptionHandlerTest.java', 'styx-common/src/test/java/com/spotify/styx/storage/DatastoreStorageTest.java']