watchman: add defer_vcs boolean option for subscriptions

Summary:
The Nuclide folks need this so that they can be smarter around rebases
and other similar operations.

Note: I still need to update the docs for this, just wanted to diff this out
early.

Test Plan:
`make integration`.  Added a new test to show that we immediately
dispatch when `.hg/wlock` is changed.

Reviewers: sid0

Reviewed By: sid0

Differential Revision: https://reviews.facebook.net/D39009

cmds/subscribe.c

@@ -173,6 +173,7 @@ static void cmd_subscribe(struct watchman_client *client, json_t *args)
   json_t *query_spec;
   struct w_query_field_list field_list;
   char *errmsg;
+  int defer = true; /* can't use bool because json_unpack requires int */
 
   if (json_array_size(args) != 4) {
     send_error_response(client, "wrong number of arguments for subscribe");
@@ -215,6 +216,10 @@ static void cmd_subscribe(struct watchman_client *client, json_t *args)
 
   sub->name = w_string_new(name);
   sub->query = query;
+
+  json_unpack(query_spec, "{s?:b}", "defer_vcs", &defer);
+  sub->vcs_defer = defer;
+
   memcpy(&sub->field_list, &field_list, sizeof(field_list));
   sub->root = root;
 

docs/cmd.subscribe.markdown

@@ -77,3 +77,35 @@ the `query` command with the `since` generator.
 The suggested mode of operation is for the client process to maintain its own
 local copy of the last "clock" value and use that to establish the subscription
 when it first connects.
+
+## Filesystem Settling
+
+Prior to watchman version 3.2, the settling behavior was to hold subscription
+notifications until the kernel notification stream was complete.
+
+Starting in watchman version 3.2, after the notification stream is complete, if
+the root appears to be a version control directory, subscription notifications
+will be held until an outstanding version control operation is complete (at the
+time of writing, this is based on the presence of either `.hg/wlock` or
+`.git/index.lock`).  This behavior matches triggers and helps to avoid
+performing transient work in response to files changing, for example, during a
+rebase operation.
+
+In some circumstances it is desirable for a client to observe the creation of
+the control files at the start of a version control operation.  You may specify
+that you want this behavior by passing the `defer_vcs` flag to your subscription
+command invocation:
+
+```bash
+$ watchman -j -p <<-EOT
+["subscribe", "/path/to/root", "mysubscriptionname", {
+  "expression": ["allof",
+    ["type", "f"],
+    ["not", "empty"],
+    ["suffix", "php"]
+  ],
+  "defer_vcs": false,
+  "fields": ["name"]
+}]
+EOT
+```

root.c

@@ -1260,28 +1260,21 @@ static bool is_vcs_op_in_progress(w_root_t *root) {
 static void process_subscriptions(w_root_t *root)
 {
   w_ht_iter_t iter;
+  bool vcs_in_progress;
 
-  if (root->last_sub_tick == root->pending_sub_tick) {
-    return;
+  pthread_mutex_lock(&w_client_lock);
+
+  if (!w_ht_first(clients, &iter)) {
+    // No subscribers
+    goto done;
   }
 
   // If it looks like we're in a repo undergoing a rebase or
   // other similar operation, we want to defer subscription
   // notifications until things settle down
-  if (is_vcs_op_in_progress(root)) {
-    w_log(W_LOG_DBG, "deferring subscription notifications "
-        "until VCS operations complete\n");
-    return;
-  }
+  vcs_in_progress = is_vcs_op_in_progress(root);
 
-  w_log(W_LOG_DBG, "sub last=%" PRIu32 "  pending=%" PRIu32 "\n",
-      root->last_sub_tick,
-      root->pending_sub_tick);
-
-  /* now look for subscribers */
-  w_log(W_LOG_DBG, "looking for connected subscribers\n");
-  pthread_mutex_lock(&w_client_lock);
-  if (w_ht_first(clients, &iter)) do {
+  do {
     struct watchman_client *client = w_ht_val_ptr(iter.value);
     w_ht_iter_t citer;
 
@@ -1290,20 +1283,31 @@ static void process_subscriptions(w_root_t *root)
     if (w_ht_first(client->subscriptions, &citer)) do {
       struct watchman_client_subscription *sub = w_ht_val_ptr(citer.value);
 
-      w_log(W_LOG_DBG, "sub=%p %s\n", sub, sub->name->buf);
       if (sub->root != root) {
         w_log(W_LOG_DBG, "root doesn't match, skipping\n");
         continue;
       }
+      w_log(W_LOG_DBG, "sub=%p %s, last=%" PRIu32 " pending=%" PRIu32 "\n",
+          sub, sub->name->buf, sub->last_sub_tick, root->pending_sub_tick);
+
+      if (sub->last_sub_tick == root->pending_sub_tick) {
+        continue;
+      }
+
+      if (sub->vcs_defer && vcs_in_progress) {
+        w_log(W_LOG_DBG, "deferring subscription notifications for %s "
+          "until VCS operations complete\n", sub->name->buf);
+        continue;
+      }
 
       w_run_subscription_rules(client, sub, root);
+      sub->last_sub_tick = root->pending_sub_tick;
 
     } while (w_ht_next(client->subscriptions, &citer));
 
   } while (w_ht_next(clients, &iter));
+done:
   pthread_mutex_unlock(&w_client_lock);
-
-  root->last_sub_tick = root->pending_sub_tick;
 }
 
 /* process any pending triggers.

tests/integration/subscribe.php

@@ -7,6 +7,70 @@ function needsLiveConnection() {
     return true;
   }
 
+  function testImmediateSubscribe() {
+    $dir = PhutilDirectoryFixture::newEmptyFixture();
+    $root = realpath($dir->getPath());
+    mkdir("$root/.hg");
+
+    $this->watch($root);
+    $this->assertFileList($root, array('.hg'));
+    try {
+      $sub = $this->watchmanCommand('subscribe', $root, 'nodefer', array(
+        'fields' => array('name', 'exists'),
+        'defer_vcs' => false,
+      ));
+
+      $this->waitForSub('nodefer', function ($data) {
+        return true;
+      });
+      list($sub) = $this->getSubData('nodefer');
+
+      $this->assertEqual(true, $sub['is_fresh_instance']);
+      $files = $sub['files'];
+      $this->assertEqual(
+        array(
+          array('name' => '.hg', 'exists' => true)
+        ), $files);
+
+      touch("$root/.hg/wlock");
+      $this->waitForSub('nodefer', function ($data) {
+        return true;
+      });
+      $sub = $this->tail($this->getSubData('nodefer'));
+      $wlock = null;
+      foreach ($sub['files'] as $ent) {
+        if ($ent['name'] == '.hg/wlock') {
+          $wlock = $ent;
+        }
+      }
+      $this->assertEqual(array('name' => '.hg/wlock', 'exists' => true), $ent);
+
+      unlink("$root/.hg/wlock");
+
+      $this->waitForSub('nodefer', function ($data) {
+        return true;
+      });
+      $sub = $this->tail($this->getSubData('nodefer'));
+
+      $wlock = null;
+      foreach ($sub['files'] as $ent) {
+        if ($ent['name'] == '.hg/wlock') {
+          $wlock = $ent;
+        }
+      }
+      $this->assertEqual(array('name' => '.hg/wlock', 'exists' => false), $ent);
+
+      $this->watchmanCommand('unsubscribe', $root, 'nodefer');
+    } catch (Exception $e) {
+      $this->watchmanCommand('unsubscribe', $root, 'nodefer');
+      throw $e;
+    }
+  }
+
+  function tail(array $array) {
+    return end($array);
+  }
+
   function testSubscribe() {
     $dir = PhutilDirectoryFixture::newEmptyFixture();
     $root = realpath($dir->getPath());
@@ -42,7 +106,7 @@ function testSubscribe() {
       $this->waitForSub('myname', function ($data) {
         return true;
       });
-      list($sub) = $this->getSubData('myname');
+      $sub = $this->tail($this->getSubData('myname'));
 
       $this->assertEqual(false, $sub['is_fresh_instance']);
       $expect = array('a/lemon');

watchman.h

@@ -405,7 +405,6 @@ struct watchman_root {
   uint32_t next_cmd_id;
   uint32_t last_trigger_tick;
   uint32_t pending_trigger_tick;
-  uint32_t last_sub_tick;
   uint32_t pending_sub_tick;
   uint32_t last_age_out_tick;
   time_t last_age_out_timestamp;
@@ -785,6 +784,8 @@ struct watchman_client_subscription {
   w_root_t *root;
   w_string_t *name;
   w_query *query;
+  bool vcs_defer;
+  uint32_t last_sub_tick;
   struct w_query_field_list field_list;
 };