Resolve watch directories outside main thread + async iterator and symlink fixes (#3846)

* linux working pending tests with FSEvents

* add more tests, fix async iterator

* remove unnecessary check

* fix macos symlink on directories

* remove indirection layer

* todos

* fixes and some permission test

* fix opsie and make prisma test more reliable

* rebase with main

* add comptime check for macOS

* oops

* oops2

* fix symlinks cascade on FSEvents

* use JSC.WorkPool

* use withResolver, createFIFO and fix close event on async iterator

* remove unused events

---------

Co-authored-by: Jarred Sumner <jarred@jarredsumner.com>

2 files changed, 41 insertions, 17 deletions

diff --git a/src/js/node/fs.promises.ts b/src/js/node/fs.promises.ts
index 0bf6eb9b2..bdbacd27d 100644
--- a/src/js/node/fs.promises.ts
+++ b/src/js/node/fs.promises.ts
@@ -2,6 +2,7 @@
 
 // Note: `constants` is injected into the top of this file
 declare var constants: typeof import("node:fs/promises").constants;
+const { createFIFO } = $lazy("primordials");
 
 var fs = Bun.fs();
 
@@ -26,7 +27,7 @@ export function watch(
     eventType: string;
     filename: string | Buffer | undefined;
   };
-  const events: Array<Event> = [];
+
   if (filename instanceof URL) {
     throw new TypeError("Watch URLs are not supported yet");
   } else if (Buffer.isBuffer(filename)) {
@@ -38,32 +39,55 @@ export function watch(
   if (typeof options === "string") {
     options = { encoding: options };
   }
-  fs.watch(filename, options || {}, (eventType: string, filename: string | Buffer | undefined) => {
-    events.push({ eventType, filename });
+  const queue = createFIFO();
+
+  const watcher = fs.watch(filename, options || {}, (eventType: string, filename: string | Buffer | undefined) => {
+    queue.push({ eventType, filename });
     if (nextEventResolve) {
       const resolve = nextEventResolve;
       nextEventResolve = null;
       resolve();
     }
   });
+
   return {
-    async *[Symbol.asyncIterator]() {
+    [Symbol.asyncIterator]() {
       let closed = false;
-      while (!closed) {
-        while (events.length) {
-          let event = events.shift() as Event;
-          if (event.eventType === "close") {
-            closed = true;
-            break;
+      return {
+        async next() {
+          while (!closed) {
+            let event: Event;
+            while ((event = queue.shift() as Event)) {
+              if (event.eventType === "close") {
+                closed = true;
+                return { value: undefined, done: true };
+              }
+              if (event.eventType === "error") {
+                closed = true;
+                throw event.filename;
+              }
+              return { value: event, done: false };
+            }
+            const { promise, resolve } = Promise.withResolvers();
+            nextEventResolve = resolve;
+            await promise;
           }
-          if (event.eventType === "error") {
+          return { value: undefined, done: true };
+        },
+
+        return() {
+          if (!closed) {
+            watcher.close();
             closed = true;
-            throw event.filename;
+            if (nextEventResolve) {
+              const resolve = nextEventResolve;
+              nextEventResolve = null;
+              resolve();
+            }
           }
-          yield event;
-        }
-        await new Promise((resolve: Function) => (nextEventResolve = resolve));
-      }
+          return { value: undefined, done: true };
+        },
+      };
     },
   };
 }
diff --git a/src/js/out/modules/node/fs.promises.js b/src/js/out/modules/node/fs.promises.js
index 852836098..df383069b 100644
--- a/src/js/out/modules/node/fs.promises.js
+++ b/src/js/out/modules/node/fs.promises.js
@@ -1 +1 @@
-var o=(S)=>{return import.meta.require(S)};function G(S,U={}){const A=[];if(S instanceof URL)throw new TypeError("Watch URLs are not supported yet");else if(Buffer.isBuffer(S))S=S.toString();else if(typeof S!=="string")throw new TypeError("Expected path to be a string or Buffer");let z=null;if(typeof U==="string")U={encoding:U};return C.watch(S,U||{},(q,g)=>{if(A.push({eventType:q,filename:g}),z){const B=z;z=null,B()}}),{async*[Symbol.asyncIterator](){let q=!1;while(!q){while(A.length){let g=A.shift();if(g.eventType==="close"){q=!0;break}if(g.eventType==="error")throw q=!0,g.filename;yield g}await new Promise((g)=>z=g)}}}}var C=Bun.fs(),D="::bunternal::",J={[D]:(S)=>{return async function(...U){return await 1,S.apply(C,U)}}}[D],H=J(C.accessSync),I=J(C.appendFileSync),K=J(C.closeSync),L=J(C.copyFileSync),M=J(C.existsSync),N=J(C.chownSync),O=J(C.chmodSync),P=J(C.fchmodSync),Q=J(C.fchownSync),V=J(C.fstatSync),X=J(C.fsyncSync),Y=J(C.ftruncateSync),Z=J(C.futimesSync),_=J(C.lchmodSync),$=J(C.lchownSync),T=J(C.linkSync),W=C.lstat.bind(C),j=J(C.mkdirSync),x=J(C.mkdtempSync),E=J(C.openSync),F=J(C.readSync),k=J(C.writeSync),R=C.readdir.bind(C),w=C.readFile.bind(C),h=J(C.writeFileSync),b=J(C.readlinkSync),u=J(C.realpathSync),d=J(C.renameSync),c=C.stat.bind(C),v=J(C.symlinkSync),a=J(C.truncateSync),l=J(C.unlinkSync),y=J(C.utimesSync),p=J(C.lutimesSync),m=J(C.rmSync),n=J(C.rmdirSync),t=(S,U,A)=>{return new Promise((z,q)=>{try{var g=C.writevSync(S,U,A)}catch(B){q(B);return}z({bytesWritten:g,buffers:U})})},r=(S,U,A)=>{return new Promise((z,q)=>{try{var g=C.readvSync(S,U,A)}catch(B){q(B);return}z({bytesRead:g,buffers:U})})},i={access:H,appendFile:I,close:K,copyFile:L,exists:M,chown:N,chmod:O,fchmod:P,fchown:Q,fstat:V,fsync:X,ftruncate:Y,futimes:Z,lchmod:_,lchown:$,link:T,lstat:W,mkdir:j,mkdtemp:x,open:E,read:F,write:k,readdir:R,readFile:w,writeFile:h,readlink:b,realpath:u,rename:d,stat:c,symlink:v,truncate:a,unlink:l,utimes:y,lutimes:p,rm:m,rmdir:n,watch:G,writev:t,readv:r,constants,[Symbol.for("CommonJS")]:0};export{t as writev,h as writeFile,k as write,G as watch,y as utimes,l as unlink,a as truncate,v as symlink,c as stat,n as rmdir,m as rm,d as rename,u as realpath,r as readv,b as readlink,R as readdir,w as readFile,F as read,E as open,x as mkdtemp,j as mkdir,p as lutimes,W as lstat,T as link,$ as lchown,_ as lchmod,Z as futimes,Y as ftruncate,X as fsync,V as fstat,Q as fchown,P as fchmod,M as exists,i as default,L as copyFile,K as close,N as chown,O as chmod,I as appendFile,H as access};
+var s=(z)=>{return import.meta.require(z)};function N(z,B={}){if(z instanceof URL)throw new TypeError("Watch URLs are not supported yet");else if(Buffer.isBuffer(z))z=z.toString();else if(typeof z!=="string")throw new TypeError("Expected path to be a string or Buffer");let C=null;if(typeof B==="string")B={encoding:B};const G=M(),H=S.watch(z,B||{},(D,A)=>{if(G.push({eventType:D,filename:A}),C){const I=C;C=null,I()}});return{[Symbol.asyncIterator](){let D=!1;return{async next(){while(!D){let A;while(A=G.shift()){if(A.eventType==="close")return D=!0,{value:void 0,done:!0};if(A.eventType==="error")throw D=!0,A.filename;return{value:A,done:!1}}const{promise:I,resolve:L}=Promise.withResolvers();C=L,await I}return{value:void 0,done:!0}},return(){if(!D){if(H.close(),D=!0,C){const A=C;C=null,A()}}return{value:void 0,done:!0}}}}}}var{createFIFO:M}=globalThis[Symbol.for("Bun.lazy")]("primordials"),S=Bun.fs(),K="::bunternal::",J={[K]:(z)=>{return async function(...B){return await 1,z.apply(S,B)}}}[K],P=J(S.accessSync),Q=J(S.appendFileSync),U=J(S.closeSync),V=J(S.copyFileSync),X=J(S.existsSync),Y=J(S.chownSync),Z=J(S.chmodSync),_=J(S.fchmodSync),$=J(S.fchownSync),q=J(S.fstatSync),O=J(S.fsyncSync),g=J(S.ftruncateSync),T=J(S.futimesSync),W=J(S.lchmodSync),j=J(S.lchownSync),k=J(S.linkSync),E=S.lstat.bind(S),h=J(S.mkdirSync),w=J(S.mkdtempSync),x=J(S.openSync),F=J(S.readSync),R=J(S.writeSync),b=S.readdir.bind(S),u=S.readFile.bind(S),d=J(S.writeFileSync),c=J(S.readlinkSync),v=J(S.realpathSync),a=J(S.renameSync),y=S.stat.bind(S),l=J(S.symlinkSync),p=J(S.truncateSync),m=J(S.unlinkSync),n=J(S.utimesSync),t=J(S.lutimesSync),r=J(S.rmSync),o=J(S.rmdirSync),f=(z,B,C)=>{return new Promise((G,H)=>{try{var D=S.writevSync(z,B,C)}catch(A){H(A);return}G({bytesWritten:D,buffers:B})})},i=(z,B,C)=>{return new Promise((G,H)=>{try{var D=S.readvSync(z,B,C)}catch(A){H(A);return}G({bytesRead:D,buffers:B})})},SS={access:P,appendFile:Q,close:U,copyFile:V,exists:X,chown:Y,chmod:Z,fchmod:_,fchown:$,fstat:q,fsync:O,ftruncate:g,futimes:T,lchmod:W,lchown:j,link:k,lstat:E,mkdir:h,mkdtemp:w,open:x,read:F,write:R,readdir:b,readFile:u,writeFile:d,readlink:c,realpath:v,rename:a,stat:y,symlink:l,truncate:p,unlink:m,utimes:n,lutimes:t,rm:r,rmdir:o,watch:N,writev:f,readv:i,constants,[Symbol.for("CommonJS")]:0};export{f as writev,d as writeFile,R as write,N as watch,n as utimes,m as unlink,p as truncate,l as symlink,y as stat,o as rmdir,r as rm,a as rename,v as realpath,i as readv,c as readlink,b as readdir,u as readFile,F as read,x as open,w as mkdtemp,h as mkdir,t as lutimes,E as lstat,k as link,j as lchown,W as lchmod,T as futimes,g as ftruncate,O as fsync,q as fstat,$ as fchown,_ as fchmod,X as exists,SS as default,V as copyFile,U as close,Y as chown,Z as chmod,Q as appendFile,P as access};