fix: don't clear context.meta on finalize, production needs that

src/bridge/proxy_stream.ts

@@ -27,29 +27,29 @@ export class ProxyStream {
     this.stream = base
     this.buffered = []
     this.bufferedByteLength = 0
-    this.stream.on("close", ()=> this.ended = true)
-    this.stream.on("end", ()=> this.ended = true)
-    this.stream.on("error", ()=> this.ended = true)
+    this.stream.on("close", () => this.ended = true)
+    this.stream.on("end", () => this.ended = true)
+    this.stream.on("error", () => this.ended = true)
   }
 
-  read(size?: number): Buffer | null{
+  read(size?: number): Buffer | null {
     let chunk = this.stream.read(size)
-    if(chunk && !this.tainted) this.bufferChunk(chunk)
+    if (chunk && !this.tainted) this.bufferChunk(chunk)
     return chunk
   }
 
-  bufferChunk(chunk: Buffer) : boolean{
-    if(this.tainted) return false
+  bufferChunk(chunk: Buffer): boolean {
+    if (this.tainted) return false
     this.bufferedByteLength += chunk.byteLength
-    if(this.bufferedByteLength > 10 * 1024 * 1024){
+    if (this.bufferedByteLength > 10 * 1024 * 1024) {
       // no longer buffering, can't be used
       this.tainted = true
       this.buffered = []
       this.bufferedByteLength = 0
-    }else{
+    } else {
       this.buffered.push(chunk)
     }
-    return !this.tainted 
+    return !this.tainted
   }
 
   get ref() {
@@ -85,25 +85,25 @@ registerBridge("subscribeProxyStream", function (ctx: Context, config: Config, r
   })*/
   //setImmediate(() => stream.resume())
 })
-registerBridge("readProxyStream", function(ctx: Context, config: Config, ref: ivm.Reference<ProxyStream>, cb: ivm.Reference<Function>){
-  ctx.addCallback(cb)
+registerBridge("readProxyStream", function (ctx: Context, config: Config, ref: ivm.Reference<ProxyStream>, cb: ivm.Reference<Function>) {
   const proxyable = ref.deref({ release: true })
   const stream = proxyable.stream
 
   let attempts = 0
-  const tryRead = function(){
+  const tryRead = function () {
     attempts += 1
     let chunk = proxyable.read(1024 * 1024)
     let data: ivm.Copy<ArrayBuffer> | null = null
-    if(chunk){
+    if (chunk) {
       data = transferInto(chunk)
     }
-    if(data || attempts >= 10 || proxyable.ended){
+    ctx.addCallback(cb)
+    if (data || attempts >= 10 || proxyable.ended) {
       ctx.applyCallback(cb, [null, data, proxyable.tainted])
-    }else if(attempts >= 10 && !proxyable.ended){
+    } else if (attempts >= 10 && !proxyable.ended) {
       // wait a bit, with a backoff
       setTimeout(tryRead, 20 * attempts)
-    }else{
+    } else {
       ctx.applyCallback(cb, [null, null, proxyable.tainted])
     }
   }

src/context.ts

@@ -68,6 +68,7 @@ export class Context extends EventEmitter {
 
 	addCallback(fn: ivm.Reference<Function>) {
 		this.callbacks.push(fn)
+		log.silly("Added callback", fn)
 		this.emit("callbackAdded", fn)
 	}
 
@@ -81,6 +82,7 @@ export class Context extends EventEmitter {
 	}
 
 	async _applyCallback(fn: ivm.Reference<Function>, args: any[], opts?: any) {
+		log.silly("Applying callback", fn, args)
 		try {
 			if (this.iso.isDisposed)
 				return
@@ -237,12 +239,13 @@ export class Context extends EventEmitter {
 				}
 				log.silly("Callbacks present initially, waiting.")
 				const cbFn = () => {
+					log.silly("Callback applied handler in finalize.", this.callbacks.length)
 					if (this.callbacks.length === 0) {
 						this.removeListener("callbackApplied", cbFn)
 						resolve()
 						return
 					}
-					log.silly("Callbacks still present, waiting.")
+					log.silly("Callbacks still present, waiting.", this.callbacks.length)
 				}
 				this.on("callbackApplied", cbFn)
 			})
@@ -260,7 +263,6 @@ export class Context extends EventEmitter {
 					// don't really care
 				}
 			}
-			this.meta = {} // reset own meta
 			this.logMetadata = {} // reset log meta data!
 		}
 	}

src/default_context_store.ts

@@ -37,7 +37,6 @@ export class DefaultContextStore implements ContextStore {
     try {
       t2 = t.start("createContext")
       const ctx = await createContext(config, iso, { inspector: !!this.options.inspect })
-      ctx.meta.iso = iso
       t2.end()
 
       ctx.set('app', app.forV8())

v8env/fly/streams.js

@@ -1,64 +1,65 @@
-export default function streamsInit(ivm, dispatcher){
+import { logger } from '../logger'
+
+export default function streamsInit(ivm, dispatcher) {
   return {
-    refToStream(ref){
+    refToStream(ref) {
       let closed = false
-      let resumed = false
       const r = new ReadableStream({
         start(controller) {
           dispatcher.dispatch("subscribeProxyStream",
             ref,
-            new ivm.Reference((name, ...args) => {
+            new ivm.Reference(function (name, ...args) {
+              logger.debug("GOT EVENT:", name)
               if (name === "close" || name === "end") {
                 if (!closed) {
                   closed = true
                   controller.close()
                 }
               } else if (name === "error") {
                 controller.error(new Error(args[0]))
-              /*} else if (name === "data") {
-                controller.enqueue(args[0])
-                if(controller.desiredSize <= 0 && resumed){
-                  resumed = false
-                  dispatcher.dispatch("controlProxyStream", "pause", ref)
-                }//*/ //not using events, calling read manually with pull for now
+                /*} else if (name === "data") {
+                  controller.enqueue(args[0])
+                  if(controller.desiredSize <= 0 && resumed){
+                    resumed = false
+                    dispatcher.dispatch("controlProxyStream", "pause", ref)
+                  }//*/ //not using events, calling read manually with pull for now
               } else
                 logger.debug("unhandled event", name)
             })
           )
         },
-        pull(controller){
+        pull(controller) {
           //if(r.locked && !resumed){
-            if(closed){
-              return Promise.resolve(null)
-            }
-            return new Promise((resolve, reject) => {
-              resumed = true
-              dispatcher.dispatch("readProxyStream", ref, new ivm.Reference((err, data, tainted) => {
-                if(err){
-                  controller.error(new Error(err))
-                  reject(err)
-                  return
-                }
-                // if data is blank the stream will keep pulling
-                // readProxyStream tries a few times to minimize bridge calls
-                controller.enqueue(data)
-                if(data && r._ref && tainted){
-                  // once underlying ref is tainted, we can't pass it around anymore
-                  // but we can still use it internally
-                  r._ref = undefined
-                }
-                resolve()
-              }))
+          if (closed) {
+            return Promise.resolve(null)
+          }
+          return new Promise((resolve, reject) => {
+            dispatcher.dispatch("readProxyStream", ref, new ivm.Reference((err, data, tainted) => {
+              if (err) {
+                controller.error(new Error(err))
+                reject(err)
+                return
+              }
+              // if data is blank the stream will keep pulling
+              // readProxyStream tries a few times to minimize bridge calls
+              controller.enqueue(data)
+              if (data && r._ref && tainted) {
+                // once underlying ref is tainted, we can't pass it around anymore
+                // but we can still use it internally
+                r._ref = undefined
+              }
+              resolve()
+            }))
 
-            })
+          })
           //}
         },
         cancel() {
           logger.debug("readable stream was cancelled")
         }
       })
       r._ref = ref
-      return r 
+      return r
     }
   }
 }