Fix bug with polling join event

Author: RainingComputers

pkg/platforms/postgres/postgres.go

@@ -1802,10 +1802,19 @@ func joins(
 			UPDATE isolates
 			SET lock = $1
 			WHERE id IN (
-				SELECT id FROM isolates
-				WHERE status = 'awaiting'
-					AND lock = -1
-					AND events @> '[{"type":"JoinEvent"}]'
+				SELECT parent.id FROM isolates parent
+				WHERE parent.status = 'awaiting'
+					AND parent.lock = -1
+					AND parent.events @> '[{"type":"JoinEvent"}]'
+					AND EXISTS (
+						SELECT 1
+						FROM jsonb_array_elements(parent.events)
+							WITH ORDINALITY AS e(elem, idx)
+						JOIN isolates child
+							ON child.name = e.elem->>'id'
+							AND child.status = 'done'
+						WHERE e.elem->>'type' = 'JoinEvent'
+					)
 				FOR UPDATE SKIP LOCKED
 				LIMIT $2
 			)

pkg/runtimes/dupher/worker_test.go

@@ -632,6 +632,41 @@ func CallerFlowSpec() {
 				},
 			},
 		},
+		{
+			name:     "flow_function_call_no_return",
+			flowName: "VoidCallerFlow",
+			source: map[string]string{
+				"voidchild.ds.go": `package main
+
+import (
+	"github.com/RainingComputers/deepslate/prelude"
+	"time"
+)
+
+func VoidChildSpec(a int, b int) {
+	result := a + b
+	prelude.Sleep(1 * time.Second)
+	_ = result
+}
+`,
+				"voidcaller.ds.go": `package main
+
+func VoidCallerFlowSpec() {
+	VoidChildSpec(3, 5)
+	var done int = 1
+	_ = done
+}
+`,
+			},
+			steps: []workerTestStep{
+				{
+					sleep: 3 * time.Second,
+					wantVars: map[string]any{
+						"done": 1,
+					},
+				},
+			},
+		},
 	}
 
 	httpAddr, controlAddr, shutdown, err := spawnBinary(tests)

pkg/runtimes/dython/deepslate/tests/test_worker.py

@@ -580,6 +580,39 @@ async def caller_flow():
             ],
             want_traces=[],
         ),
+        WorkerTestCase(
+            name="flow_function_call_no_return",
+            flow_name="caller_flow",
+            source={
+                "noop.py": """\
+from deepslate import flow, sleep
+
+
+@flow
+async def noop(a: int, b: int) -> None:
+    result: int = a + b
+    sleep(1.0)
+""",
+                "caller_flow.py": """\
+from deepslate import flow
+from .noop import noop
+
+
+@flow
+async def caller_flow():
+    finished: int = 0
+    noop(3, 5)
+    finished = 1
+""",
+            },
+            steps=[
+                WorkerTestStep(
+                    sleep=3.0,
+                    want_vars={"finished": 1},
+                ),
+            ],
+            want_traces=[],
+        ),
     ],
     ids=lambda t: t.name,
 )