Restore and improve cluster run command

api/process.capnp

@@ -9,13 +9,13 @@ $Go.import("github.com/wetware/pkg/api/process");
 interface Executor {
     # Executor has the ability to create and run WASM processes given the
     # WASM bytecode.
-    exec @0 (bytecode :Data, bootstrapClient :Capability) -> (process :Process);
+    exec @0 (bytecode :Data, ppid :UInt32, bootstrapClient :Capability) -> (process :Process);
     # Exec creates an runs a process from the provided bytecode. Optionally, a
     # capability can be passed through the `cap` parameter. This capability will
     # be available at the process bootContext.
     #
     # The Process capability is associated to the created process.
-    execCached @1 (cid :Text, bootstrapClient :Capability) -> (process :Process);
+    execCached @1 (cid :Text, ppid :UInt32, bootstrapClient :Capability) -> (process :Process);
     # Same as Exec, but the bytecode is directly from the BytecodeRegistry.
     # Provides a significant performance improvement for medium to large
     # WASM streams.

cap/csp/executor.go

@@ -35,9 +35,33 @@ func (ex Executor) Release() {
 	capnp.Client(ex).Release()
 }
 
-func (ex Executor) Exec(ctx context.Context, src []byte) (Proc, capnp.ReleaseFunc) {
-	f, release := api.Executor(ex).Exec(ctx, func(ps api.Executor_exec_Params) error {
-		return ps.SetBytecode(src)
-	})
+// Exec spawns a new process from WASM bytecode bc. If the caller is a WASM process
+// spawned in this same executor, it should use its PID as ppid to mark the
+// new process as a subprocess.
+func (ex Executor) Exec(ctx context.Context, bc []byte, ppid uint32, client capnp.Client) (Proc, capnp.ReleaseFunc) {
+	f, release := api.Executor(ex).Exec(ctx,
+		func(ps api.Executor_exec_Params) error {
+			if err := ps.SetBytecode(bc); err != nil {
+				return err
+			}
+
+			ps.SetPpid(ppid)
+			return ps.SetBootstrapClient(client)
+		})
+	return Proc(f.Process()), release
+}
+
+// ExecFromCache behaves the same way as Exec, but expects the bytecode to be already
+// cached at the executor.
+func (ex Executor) ExecFromCache(ctx context.Context, cid string, ppid uint32, client capnp.Client) (Proc, capnp.ReleaseFunc) {
+	f, release := api.Executor(ex).ExecCached(ctx,
+		func(ps api.Executor_execCached_Params) error {
+			if err := ps.SetCid(cid); err != nil {
+				return err
+			}
+
+			ps.SetPpid(ppid)
+			return ps.SetBootstrapClient(client)
+		})
 	return Proc(f.Process()), release
 }

cap/csp/proc.go → cap/csp/process.go

@@ -25,12 +25,26 @@ func (p Proc) Release() {
 	capnp.Client(p).Release()
 }
 
-// func (p Proc) Kill(ctx context.Context) error {
-// 	f, release := api.Process(p).Kill(ctx, nil)
-// 	defer release()
+// Kill a process and any sub processes it might have spawned.
+func (p Proc) Kill(ctx context.Context) error {
+	f, release := api.Process(p).Kill(ctx, nil)
+	defer release()
+
+	select {
+	case <-f.Done():
+	case <-ctx.Done():
+	}
+
+	if ctx.Err() != nil {
+		return ctx.Err()
+	}
 
-// 	return casm.Future(f).Await(ctx)
-// }
+	_, err := f.Struct()
+	if err != nil {
+		return err
+	}
+	return nil
+}
 
 func (p Proc) Wait(ctx context.Context) error {
 	f, release := api.Process(p).Wait(ctx, nil)