changes is less likely to miss updates, and changes test is more robust

git-svn-id: https://svn.apache.org/repos/asf/couchdb/trunk@931407 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 64 insertions, 23 deletions

diff --git a/share/www/script/test/changes.js b/share/www/script/test/changes.js
index 226204ef..1240712f 100644
--- a/share/www/script/test/changes.js
+++ b/share/www/script/test/changes.js
@@ -17,7 +17,7 @@ function jsonp(obj) {
 }
 
 couchTests.changes = function(debug) {
-  var db = new CouchDB("test_suite_db", {"X-Couch-Full-Commit":"false"});
+  var db = new CouchDB("test_suite_db", {"X-Couch-Full-Commit":"true"});
   db.deleteDb();
   db.createDb();
   if (debug) debugger;
@@ -29,11 +29,13 @@ couchTests.changes = function(debug) {
   var docFoo = {_id:"foo", bar:1};
   T(db.save(docFoo).ok);
   T(db.ensureFullCommit().ok);
+  T(db.open(docFoo._id)._id == docFoo._id);
   
   req = CouchDB.request("GET", "/test_suite_db/_changes");
   var resp = JSON.parse(req.responseText);
 
-  T(resp.results.length == 1 && resp.last_seq==1, "one doc db")
+  T(resp.last_seq == 1);
+  T(resp.results.length == 1, "one doc db")
   T(resp.results[0].changes[0].rev == docFoo._rev)
 
   // test with callback
@@ -70,7 +72,7 @@ couchTests.changes = function(debug) {
     // WebKit (last checked on nightly #47686) does fail on processing
     // the async-request properly while javascript is executed.
 
-    xhr.open("GET", "/test_suite_db/_changes?feed=continuous", true);
+    xhr.open("GET", "/test_suite_db/_changes?feed=continuous&timeout=500", true);
     xhr.send("");
 
     var docBar = {_id:"bar", bar:1};
@@ -108,7 +110,7 @@ couchTests.changes = function(debug) {
     xhr = CouchDB.newXhr();
 
     //verify the hearbeat newlines are sent
-    xhr.open("GET", "/test_suite_db/_changes?feed=continuous&heartbeat=10", true);
+    xhr.open("GET", "/test_suite_db/_changes?feed=continuous&heartbeat=10&timeout=500", true);
     xhr.send("");
     
     var str;
@@ -122,6 +124,8 @@ couchTests.changes = function(debug) {
     T(str.charAt(str.length - 1) == "\n")
     T(str.charAt(str.length - 2) == "\n")
 
+    // otherwise we'll continue to receive heartbeats forever
+    xhr.abort();
 
     // test longpolling
     xhr = CouchDB.newXhr();
@@ -180,6 +184,14 @@ couchTests.changes = function(debug) {
         return doc.user && (doc.user == req.userCtx.name);
       }),
       "conflicted" : "function(doc, req) { return (doc._conflicts);}",
+    },
+    options : {
+      local_seq : true
+    },
+    views : {
+      local_seq : {
+        map : "function(doc) {emit(doc._local_seq, null)}"
+      }
     }
   }
 
@@ -194,7 +206,7 @@ couchTests.changes = function(debug) {
   
   var req = CouchDB.request("GET", "/test_suite_db/_changes?filter=changes_filter/bop");
   var resp = JSON.parse(req.responseText);
-  T(resp.results.length == 1);
+  T(resp.results.length == 1, "filtered/bop");
     
   req = CouchDB.request("GET", "/test_suite_db/_changes?filter=changes_filter/dynamic&field=woox");
   resp = JSON.parse(req.responseText);
@@ -225,21 +237,46 @@ couchTests.changes = function(debug) {
     
     T(resp.last_seq == 9);
     T(resp.results && resp.results.length > 0 && resp.results[0]["id"] == "bingo", "filter the correct update");
-
-    // filter with continuous
-    xhr = CouchDB.newXhr();
-    xhr.open("GET", "/test_suite_db/_changes?feed=continuous&filter=changes_filter/bop&timeout=200", true);
-    xhr.send("");
-    db.save({"_id":"rusty", "bop" : "plankton"});
+    xhr.abort();
     
-    waitForSuccess(function() {
+    timeout = 500;
+    last_seq = 10
+    while (true) {
+
+      // filter with continuous
+      xhr = CouchDB.newXhr();
+      xhr.open("GET", "/test_suite_db/_changes?feed=continuous&filter=changes_filter/bop&timeout="+timeout, true);
+      xhr.send("");
+
+      db.save({"_id":"rusty", "bop" : "plankton"});
+      T(xhr.readyState != 4, "test client too slow");
+      var rusty = db.open("rusty", {cache_bust : new Date()});
+      T(rusty._id == "rusty");
+
+      waitForSuccess(function() { // throws an error after 5 seconds
+        if (xhr.readyState != 4) {
+          throw("still waiting")
+        }
+      }, "continuous-rusty");
       lines = xhr.responseText.split("\n");
-      JSON.parse(lines[3]);
-    }, "continuous-timeout");
-    
-    T(JSON.parse(lines[1]).id == "bingo", lines[1]);
-    T(JSON.parse(lines[2]).id == "rusty", lines[2]);
-    T(JSON.parse(lines[3]).last_seq == 10, lines[3]);
+      try {
+        JSON.parse(lines[3])
+        good = true;
+      } catch(e) {
+        good = false;
+      }
+      if (good) {
+        T(JSON.parse(lines[1]).id == "bingo", lines[1]);
+        T(JSON.parse(lines[2]).id == "rusty", lines[2]);
+        T(JSON.parse(lines[3]).last_seq == last_seq, lines[3]);
+        break;
+      } else {
+        xhr.abort();
+        db.deleteDoc(rusty);
+        timeout = timeout * 2;
+        last_seq = last_seq + 2;
+      }
+    }
   }
   // error conditions
 
diff --git a/share/www/script/test/uuids.js b/share/www/script/test/uuids.js
index 4de7ce90..35ac2683 100644
--- a/share/www/script/test/uuids.js
+++ b/share/www/script/test/uuids.js
@@ -97,7 +97,7 @@ couchTests.uuids = function(debug) {
       T(result.uuids[i].length == 32);
       var u1 = result.uuids[i-1].substr(0, 13);
       var u2 = result.uuids[i].substr(0, 13);
-      T(u1 < u2, "UTC uuids are roughly ordered.");
+      T(u1 < u2, "UTC uuids are only roughly ordered, so this assertion may fail occasionally. Don't sweat it.");
     }
   };
 
diff --git a/src/couchdb/couch_changes.erl b/src/couchdb/couch_changes.erl
index 35ea147f..9496c226 100644
--- a/src/couchdb/couch_changes.erl
+++ b/src/couchdb/couch_changes.erl
@@ -17,7 +17,7 @@
 
 %% @type Req -> #httpd{} | {json_req, JsonObj()}
 handle_changes(#changes_args{}=Args1, Req, Db) ->
-    Args = Args1#changes_args{filter=make_filter_fun(Args1, Req, Db)},
+    Args = Args1#changes_args{filter=make_filter_fun(Args1#changes_args.filter, Req, Db)},
     StartSeq = case Args#changes_args.dir of
     rev ->
         couch_db:get_update_seq(Db);
@@ -27,7 +27,6 @@ handle_changes(#changes_args{}=Args1, Req, Db) ->
     if Args#changes_args.feed == "continuous" orelse
         Args#changes_args.feed == "longpoll" ->
         fun(Callback) ->
-            start_sending_changes(Callback, Args#changes_args.feed),
             Self = self(),
             {ok, Notify} = couch_db_update_notifier:start_link(
                 fun({_, DbName}) when DbName == Db#db.name ->
@@ -36,6 +35,7 @@ handle_changes(#changes_args{}=Args1, Req, Db) ->
                     ok
                 end
             ),
+            start_sending_changes(Callback, Args#changes_args.feed),
             {Timeout, TimeoutFun} = get_changes_timeout(Args, Callback),
             couch_stats_collector:track_process_count(
                 Self,
@@ -72,7 +72,7 @@ handle_changes(#changes_args{}=Args1, Req, Db) ->
     end.
 
 %% @type Req -> #httpd{} | {json_req, JsonObj()}
-make_filter_fun(#changes_args{filter=FilterName}, Req, Db) ->
+make_filter_fun(FilterName, Req, Db) ->
     case [list_to_binary(couch_httpd:unquote(Part))
             || Part <- string:tokens(FilterName, "/")] of
     [] ->
@@ -94,6 +94,7 @@ make_filter_fun(#changes_args{filter=FilterName}, Req, Db) ->
             {ok, Passes} = couch_query_servers:filter_docs(
                 Req, Db, DDoc, FName, Docs
             ),
+            % ?LOG_INFO("filtering ~p ~w",[FName, [DI#doc_info.high_seq || DI <- DocInfos]]),
             [{[{<<"rev">>, couch_doc:rev_to_str(Rev)}]}
                 || #doc_info{revs=[#rev_info{rev=Rev}|_]} <- DocInfos,
                 Pass <- Passes, Pass == true]
@@ -155,20 +156,22 @@ send_changes(Args, Callback, Db, StartSeq, Prepend) ->
 
 keep_sending_changes(Args, Callback, Db, StartSeq, Prepend, Timeout,
     TimeoutFun) ->
-
     #changes_args{
         feed = ResponseType,
         limit = Limit
     } = Args,
+    % ?LOG_INFO("send_changes start ~p",[StartSeq]),
     {ok, {_, EndSeq, Prepend2, _, _, _, NewLimit, _}} = send_changes(
         Args#changes_args{dir=fwd}, Callback, Db, StartSeq, Prepend
     ),
+    % ?LOG_INFO("send_changes last ~p",[EndSeq]),
     couch_db:close(Db),
     if Limit > NewLimit, ResponseType == "longpoll" ->
         end_sending_changes(Callback, EndSeq, ResponseType);
     true ->
         case wait_db_updated(Timeout, TimeoutFun) of
         updated ->
+            % ?LOG_INFO("wait_db_updated updated ~p",[{Db#db.name, EndSeq}]),
             case couch_db:open(Db#db.name, [{user_ctx, Db#db.user_ctx}]) of
             {ok, Db2} ->
                 keep_sending_changes(
@@ -184,6 +187,7 @@ keep_sending_changes(Args, Callback, Db, StartSeq, Prepend, Timeout,
                 end_sending_changes(Callback, EndSeq, ResponseType)
             end;
         stop ->
+            % ?LOG_INFO("wait_db_updated stop ~p",[{Db#db.name, EndSeq}]),
             end_sending_changes(Callback, EndSeq, ResponseType)
         end
     end.