[tau-cli] run functions with no dream (#421)

Author: samyfodil

.gitignore

@@ -92,8 +92,8 @@ develop-eggs/
 downloads/
 eggs/
 .eggs/
-lib/
-lib64/
+/lib/
+/lib64/
 parts/
 sdist/
 var/

core/vm/config.go

@@ -11,4 +11,5 @@ type OutputType uint32
 const (
 	Pipe OutputType = iota
 	Buffer
+	Stdio
 )

pkg/builder/methods.go

@@ -100,14 +100,13 @@ func (b *builder) run(output *output, image *ci.DockerImage, environment specs.E
 			return b.Errorf("instantiating container failed with: %w", err)
 		}
 
-		log, err := container.Run(b.context)
+		log, runErr := container.Run(b.context)
 		if log != nil {
-			_, err = io.Copy(b.output, log.Combined())
-			if err != nil {
-				return b.Errorf("writting container output failed with: %w", err)
+			if _, copyErr := io.Copy(b.output, log.Combined()); copyErr != nil {
+				return b.Errorf("writing container output failed with: %w", copyErr)
 			}
 		}
-		if err != nil {
+		if runErr != nil {
 			json.NewEncoder(b.output).Encode(struct {
 				Step      string `json:"step"`
 				Timestamp int64  `json:"timestamp"`
@@ -117,9 +116,9 @@ func (b *builder) run(output *output, image *ci.DockerImage, environment specs.E
 				Step:      script,
 				Timestamp: time.Now().UnixNano(),
 				Status:    "error",
-				Error:     err.Error(),
+				Error:     runErr.Error(),
 			})
-			return b.Errorf("running container failed with: %w", err)
+			return b.Errorf("running container failed with: %w", runErr)
 		}
 		json.NewEncoder(b.output).Encode(struct {
 			Step      string `json:"step"`

pkg/builder/new.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"os"
 
+	"github.com/jedib0t/go-pretty/v6/table"
 	iface "github.com/taubyte/tau/core/builders"
 	ci "github.com/taubyte/tau/pkg/containers"
 	"github.com/taubyte/tau/pkg/specs/builders"
@@ -58,8 +59,17 @@ func New(ctx context.Context, output io.Writer, workDir string) (iface.Builder,
 	}
 
 	env := b.config.HandleDepreciatedEnvironment()
-	fmt.Fprintf(b.output, "tau build: workDir=%s config=%s image=%s workflow=%v\n",
-		b.wd.String(), b.wd.ConfigFile(), env.Image, b.config.Workflow)
+	t := table.NewWriter()
+	t.SetOutputMirror(b.output)
+	t.SetStyle(table.StyleLight)
+	t.AppendRows([]table.Row{
+		{"Working Directory", b.wd.String()},
+		{"Config File", b.wd.ConfigFile()},
+		{"Image", env.Image},
+		{"Workflow", fmt.Sprint(b.config.Workflow)},
+	})
+	t.Render()
+	fmt.Fprintln(b.output)
 
 	// set tarball if any
 	return b, b.setTarball()

pkg/containers/container.go

@@ -6,6 +6,8 @@ import (
 )
 
 // Run starts the container and waits for the container to exit before returning the container logs.
+// Logs are returned even when the container exits with a non-zero status so callers
+// can inspect build output (e.g. compiler errors).
 func (c *Container) Run(ctx context.Context) (*MuxedReadCloser, error) {
 	imageName := ""
 	if c.image != nil {
@@ -16,28 +18,31 @@ func (c *Container) Run(ctx context.Context) (*MuxedReadCloser, error) {
 		return nil, errorContainerStart(c.id, imageName, err)
 	}
 
-	if err := c.Wait(ctx); err != nil {
-		return nil, err
-	}
+	waitErr := c.Wait(ctx)
 
 	info, err := c.backend.Inspect(ctx, c.id)
 	if err != nil {
+		if waitErr != nil {
+			return nil, waitErr
+		}
 		return nil, errorContainerInspect(c.id, imageName, err)
 	}
 
-	var RetCodeErr error
-	if info.ExitCode != 0 {
-		RetCodeErr = errorContainerExitCode(c.id, imageName, info.ExitCode)
+	if waitErr == nil && info.ExitCode != 0 {
+		waitErr = errorContainerExitCode(c.id, imageName, info.ExitCode)
 	}
 
 	muxed, err := c.backend.Logs(ctx, c.id)
 	if err != nil {
+		if waitErr != nil {
+			return nil, waitErr
+		}
 		return nil, errorContainerLogs(c.id, imageName, err)
 	}
 
 	c.Cleanup(ctx)
 
-	return &MuxedReadCloser{reader: muxed}, RetCodeErr
+	return &MuxedReadCloser{reader: muxed}, waitErr
 }
 
 // Wait calls the ContainerWait method for the container, and returns once a response has been received.

pkg/vm/backend/file/backend.go

@@ -3,16 +3,24 @@
 package file
 
 import (
+	"archive/zip"
+	"compress/gzip"
+	"fmt"
 	"io"
 	"os"
 
+	"github.com/h2non/filetype"
+	"github.com/h2non/filetype/matchers"
 	"github.com/taubyte/tau/core/vm"
+	"github.com/taubyte/tau/pkg/specs/builders/wasm"
 	"github.com/taubyte/tau/pkg/vm/backend/errors"
 
 	ma "github.com/multiformats/go-multiaddr"
 	resolv "github.com/taubyte/tau/pkg/vm/resolvers/taubyte"
 )
 
+const headerSize = 512
+
 type backend struct{}
 
 func New() vm.Backend {
@@ -27,6 +35,28 @@ func (b *backend) Scheme() string {
 	return resolv.FILE_PROTOCOL_NAME
 }
 
+// zipEntryReadCloser closes both the zip entry reader and the underlying file.
+type zipEntryReadCloser struct {
+	io.ReadCloser
+	file *os.File
+}
+
+func (z *zipEntryReadCloser) Close() error {
+	z.ReadCloser.Close()
+	return z.file.Close()
+}
+
+// gzipFileReadCloser closes both the gzip reader and the underlying file.
+type gzipFileReadCloser struct {
+	*gzip.Reader
+	file *os.File
+}
+
+func (g *gzipFileReadCloser) Close() error {
+	g.Reader.Close()
+	return g.file.Close()
+}
+
 func (b *backend) Get(multiAddr ma.Multiaddr) (io.ReadCloser, error) {
 	protocols := multiAddr.Protocols()
 	if protocols[0].Code != resolv.P_FILE {
@@ -38,13 +68,61 @@ func (b *backend) Get(multiAddr ma.Multiaddr) (io.ReadCloser, error) {
 		return nil, errors.ParseProtocol(resolv.FILE_PROTOCOL_NAME, err)
 	}
 
-	// remove extra slash
 	path = path[1:]
 
 	file, err := os.Open(path)
 	if err != nil {
 		return nil, errors.RetrieveError(path, err, b)
 	}
 
-	return file, nil
+	header := make([]byte, headerSize)
+	n, err := file.Read(header)
+	if err != nil && err != io.EOF {
+		file.Close()
+		return nil, fmt.Errorf("reading file header: %w", err)
+	}
+	header = header[:n]
+
+	if _, err := file.Seek(0, io.SeekStart); err != nil {
+		file.Close()
+		return nil, fmt.Errorf("seeking to start: %w", err)
+	}
+
+	kind, err := filetype.Match(header)
+	if err != nil {
+		file.Close()
+		return nil, fmt.Errorf("matching file type: %w", err)
+	}
+
+	switch kind {
+	case matchers.TypeZip:
+		info, err := file.Stat()
+		if err != nil {
+			file.Close()
+			return nil, fmt.Errorf("stating file: %w", err)
+		}
+		zr, err := zip.NewReader(file, info.Size())
+		if err != nil {
+			file.Close()
+			return nil, fmt.Errorf("opening zip: %w", err)
+		}
+		entryReader, err := zr.Open(wasm.WasmFile)
+		if err != nil {
+			entryReader, err = zr.Open(wasm.DeprecatedWasmFile)
+			if err != nil {
+				file.Close()
+				return nil, fmt.Errorf("zip has no %q or %q: %w", wasm.WasmFile, wasm.DeprecatedWasmFile, err)
+			}
+		}
+		return &zipEntryReadCloser{ReadCloser: entryReader, file: file}, nil
+	case matchers.TypeGz:
+		gzReader, err := gzip.NewReader(file)
+		if err != nil {
+			file.Close()
+			return nil, fmt.Errorf("opening gzip: %w", err)
+		}
+		return &gzipFileReadCloser{Reader: gzReader, file: file}, nil
+	default:
+		return file, nil
+	}
 }

pkg/vm/resolvers/file/resolver.go

@@ -0,0 +1,28 @@
+package file
+
+import (
+	"path/filepath"
+
+	ma "github.com/multiformats/go-multiaddr"
+
+	"github.com/taubyte/tau/core/vm"
+	_ "github.com/taubyte/tau/pkg/vm/resolvers/taubyte"
+)
+
+type resolver struct {
+	wasmPath string
+}
+
+// New returns a vm.Resolver that maps any module name to a file multiaddr for the given WASM path.
+// wasmPath is resolved to an absolute path so it works regardless of working directory.
+func New(wasmPath string) vm.Resolver {
+	abs, err := filepath.Abs(wasmPath)
+	if err != nil {
+		abs = wasmPath
+	}
+	return &resolver{wasmPath: abs}
+}
+
+func (r *resolver) Lookup(ctx vm.Context, name string) (ma.Multiaddr, error) {
+	return ma.NewMultiaddr("/file/" + r.wasmPath)
+}

pkg/vm/service/wazero/output.go

@@ -27,7 +27,7 @@ func (b *buffer) Read(p []byte) (n int, err error) {
 
 func (b *buffer) Write(p []byte) (n int, err error) {
 	if b.buffer != nil {
-		b.buffer.Write(p)
+		return b.buffer.Write(p)
 	}
 
 	return 0, errors.New("buffer is closed")
@@ -96,3 +96,25 @@ func (p *pipe) Close() error {
 }
 
 var MaxOutputCapacity = 10 * 1024
+
+// stdioWriter forwards writes to a standard stream (os.Stdout or os.Stderr).
+// Read returns nothing; Close is a no-op so the process stdio is not closed.
+type stdioWriter struct {
+	*os.File
+}
+
+func (s *stdioWriter) Read(p []byte) (n int, err error) {
+	return 0, io.EOF
+}
+
+func (s *stdioWriter) Close() error {
+	return nil
+}
+
+func newStdout() io.ReadWriteCloser {
+	return &stdioWriter{File: os.Stdout}
+}
+
+func newStderr() io.ReadWriteCloser {
+	return &stdioWriter{File: os.Stderr}
+}

pkg/vm/service/wazero/runtime.go

@@ -154,7 +154,14 @@ func (r *runtime) instantiate(name string, compiled wazero.CompiledModule, hasRe
 		return nil, fmt.Errorf("instantiating compiled module `%s` failed with: %s", name, err)
 	}
 
-	if _start := m.ExportedFunction("_start"); _start != nil {
+	// reactor modules export _initialize to set up globals (e.g. os.Stdout).
+	// Must be called before any other exported function.
+	if _initialize := m.ExportedFunction("_initialize"); _initialize != nil {
+		if _, err := _initialize.Call(ctx); err != nil {
+			return nil, fmt.Errorf("calling _initialize for module `%s`: %w", name, err)
+		}
+		// TODO: this should be deleted later as we should only support reactor modules
+	} else if _start := m.ExportedFunction("_start"); _start != nil {
 		if hasReady {
 			go func() {
 				_start.Call(ctx)

pkg/vm/service/wazero/service.go

@@ -18,6 +18,9 @@ func (s *service) New(ctx vm.Context, config vm.Config) (vm.Instance, error) {
 	case vm.Buffer:
 		r.output = newBuffer()
 		r.outputErr = newBuffer()
+	case vm.Stdio:
+		r.output = newStdout()
+		r.outputErr = newStderr()
 	default:
 		var err error
 		if r.output, err = newPipe(); err != nil {