Check a pushed view that computes gets the newest value of writes made meanwhile, and queue=N's busy and its places coming back (020)

2 files changed, 44 insertions, 9 deletions

diff --git a/docs/research/122-views-off-the-connection-threads.md b/docs/research/122-views-off-the-connection-threads.md
index 5af9c09..63a1d8f 100644
--- a/docs/research/122-views-off-the-connection-threads.md
+++ b/docs/research/122-views-off-the-connection-threads.md
@@ -67,6 +67,9 @@ tail 1`, `tail 2`), 같은 것을 `--view-workers 0`으로 두 번(`control`). �
   모듈로 바꾸고(SIGHUP), 이어서 앱을 뺀다. 모든 답이 옛 모듈의 값, 새 모듈의 값, 503 `stopped`, 404 가운데 하나다(섞인 답은
   `7`로 보였을 것이다). 그 보기의 스트림이 같은 스트림에서 새 모듈의 값을 받고, 앱이 빠지면 끝난다. 값이 오기 전에 닫은 스트림
   열 개가 아무것도 남기지 않는다(멈출 때 구독자 0). `--view-workers 0`이면 보기의 스레드가 없고 같은 답이다.
+- 같은 검사에서, 스트림의 보기를 계산하는 동안 쓰기 셋이 오면 그 뒤에 마지막 값이 온다(`id`는 늘기만 한다). 그 길(무리의 `again`)을
+  끄면 스트림이 마지막 값을 받지 못해 검사가 멈춘다. 앱에 `queue=2`, 보기의 스레드 하나에 8연결이 몰리면 503 `busy`가 나오고(36번),
+  그 뒤의 `GET`은 200이다(넘긴 일의 자리가 돌아온다).
 - `server_view_bodies`(64연결이 동시에 묻는 `distance` 1,280개가 모두 모델과 같다)와 `server_push`(계산하는 보기를 미는 것 포함)가
   이제 보기의 스레드를 거쳐 그대로 초록이다.
 - Pure의 `make check`, aup와 Vega의 모든 검사가 초록이다.
diff --git a/tests/check.py b/tests/check.py
index 7d5648b..d62bb91 100644
--- a/tests/check.py
+++ b/tests/check.py
@@ -4513,9 +4513,11 @@ def server_view_workers(aup, pure, work):
     while 8 connections ask a view that loops 3,000,000 times, its app's
     module changes (SIGHUP) to one with another view in its place and the
     view tenfold, then the app is removed: every answer is the old module's,
-    the new one's, 503 stopped or 404 (gone), none mixed. Streams of it
-    opened and closed at once leave nothing behind. With --view-workers 0
-    the connection threads show it, as before."""
+    the new one's, 503 stopped or 404 (gone), none mixed. A stream of it
+    gets the last value of writes made while it is evaluated (ids only
+    growing); streams opened and closed at once leave nothing behind. Past
+    the app's queue=2 asks are 503 busy, and the places come back. With
+    --view-workers 0 the connection threads show it, as before."""
     import select
     import signal
     import threading
@@ -4547,7 +4549,15 @@ def server_view_workers(aup, pure, work):
             assert stream.head()[0] == 200
             stream.close()
         stream = EventStream(host.port, '/s/views/slow')
-        assert stream.next()[2] == '3000001'
+        first, _, value = stream.next()
+        assert value == '3000001'
+        for _ in range(3):  # written while it is evaluated: the newest snapshot's value after it
+            assert http(host.port, 'POST', '/s/inputs/bump', b'x')[0] == 200
+        ids = [first]
+        while value != '3000004':
+            ident, _, value = stream.next(timeout=10)
+            ids.append(ident)
+        assert ids == sorted(set(ids)), ids
 
         def ask():
             import http.client
@@ -4563,7 +4573,7 @@ def server_view_workers(aup, pure, work):
                     c = http.client.HTTPConnection('127.0.0.1', host.port, timeout=30)
                     continue
                 answers[got] = answers.get(got, 0) + 1
-                if got not in ((200, 'none', b'3000001'), (200, 'none', b'3000010'), (503, 'stopped', b'')) and got[0] != 404:
+                if got not in ((200, 'none', b'3000004'), (200, 'none', b'3000040'), (503, 'stopped', b'')) and got[0] != 404:
                     wrong.append(got)
             c.close()
 
@@ -4573,7 +4583,7 @@ def server_view_workers(aup, pure, work):
         time.sleep(0.5)
         write(tenfold)
         assert reading()['apps'][0]['did'] == 'changed'
-        assert stream.next(timeout=10)[2] == '3000010'  # the stream, on the new module's view
+        assert stream.next(timeout=10)[2] == '3000040'  # the stream, on the new module's view
         time.sleep(0.5)
         write(tenfold, 'q')  # s removed
         reading()
@@ -4586,7 +4596,7 @@ def server_view_workers(aup, pure, work):
     finally:
         stop.set()
         stats = host.stop()
-    old, new = answers.get((200, 'none', b'3000001'), 0), answers.get((200, 'none', b'3000010'), 0)
+    old, new = answers.get((200, 'none', b'3000004'), 0), answers.get((200, 'none', b'3000040'), 0)
     assert not wrong and old and new, (wrong[:5], answers)
     assert stats[0]['view_workers'] == 2 and stats[0]['views_by_workers'] >= old + new and stats[0]['subscribers'] == 0, stats[0]
     by_workers = stats[0]['views_by_workers']
@@ -4594,12 +4604,34 @@ def server_view_workers(aup, pure, work):
     host = HttpHost([pure / 'build/pure-serve', '--apps', conf, '--view-workers', 0])
     try:
         assert http(host.port, 'POST', '/s/inputs/bump', b'x')[0] == 200
-        assert http(host.port, 'GET', '/s/views/slow')[3] == b'3000002'  # its store kept the first bump
+        assert http(host.port, 'GET', '/s/views/slow')[3] == b'3000005'  # its store kept the bumps
     finally:
         stats = host.stop()
     assert stats[0]['view_workers'] == 0 and stats[0]['views_by_workers'] == 0, stats[0]
+    conf.write_text(f"s {module} {work / 'slow-s'} queue=2\n")
+    host = HttpHost([pure / 'build/pure-serve', '--apps', conf, '--threads', 1, '--view-workers', 1])
+    busy, other = [0], []
+    try:
+        def crowd():
+            for _ in range(6):
+                got = http(host.port, 'GET', '/s/views/slow')
+                if got[0] == 503 and got[1] == 'busy':
+                    busy[0] += 1
+                elif got[0] != 200 or got[3] != b'3000005':
+                    other.append((got[0], got[1], got[3][:20]))
+
+        threads = [threading.Thread(target=crowd) for _ in range(8)]
+        for t in threads:
+            t.start()
+        for t in threads:
+            t.join()
+        assert busy[0] and not other, (busy, other[:5])
+        assert http(host.port, 'GET', '/s/views/slow')[3] == b'3000005'  # the queue's places came back
+    finally:
+        host.stop()
     return {'old': old, 'new': new, 'stopped': answers.get((503, 'stopped', b''), 0),
-            'gone': sum(v for k, v in answers.items() if k[0] == 404), 'by_workers': by_workers}
+            'gone': sum(v for k, v in answers.items() if k[0] == 404), 'by_workers': by_workers, 'pushed_ids': len(ids),
+            'busy': busy[0]}
 
 
 PUSH_SOURCE = """record Row {