fs: add per-operation fs diagnostics channels

Add built-in node:diagnostics_channel channels for file system
operations performed through node:fs and node:fs/promises. Each
operation gets its own TracingChannel family named fs.<operation>,
with channels tracing:fs.<operation>:start, :end, :asyncStart,
:asyncEnd, and :error. The event payload carries the API
(sync/callback/promise), path/dest/fd fields when applicable, plus
result/error following TracingChannel conventions.

Events are published from the internal shared file system layer rather
than the JS wrappers, so captured function references still emit events.

Signed-off-by: Matteo Collina <hello@matteocollina.com>

Author: mcollina

doc/api/diagnostics_channel.md

@@ -1562,6 +1562,87 @@ passed to `console.warn()`.
 Emitted when `console.error()` is called. Receives and array of the arguments
 passed to `console.error()`.
 
+#### Filesystem
+
+> Stability: 1 - Experimental
+
+These channels are emitted for file system operations performed through
+`node:fs` and `node:fs/promises`. Each operation has its own
+[`TracingChannel`][] family named `fs.<operation>`, where `<operation>` is a
+stable operation name such as `open`, `read`, `write`, `stat`, `readdir`, or
+`realpath`. Subscribers can use [`diagnostics_channel.tracingChannel()`][] to
+subscribe to all events of a given operation at once:
+
+```mjs
+import diagnostics_channel from 'node:diagnostics_channel';
+
+const channel = diagnostics_channel.tracingChannel('fs.open');
+channel.subscribe({
+  start: (event) => console.log('start', event),
+  end: (event) => console.log('end', event),
+  error: (event) => console.log('error', event),
+});
+```
+
+The events are published from the internal file system implementation, so they
+are observed for every public `fs` operation regardless of whether the
+function reference was captured before subscribing or whether the operation
+uses the callback, promise, or synchronous API.
+
+Each event carries an object with the following common fields:
+
+* `api` {string} The API that performed the operation: `'sync'`, `'callback'`,
+  or `'promise'`.
+* `path` {string|undefined} The path argument for path-based operations, or
+  the source path for operations with a destination.
+* `dest` {string|undefined} The destination argument for operations that
+  accept one, such as `rename`, `link`, `symlink`, or `copyFile`.
+* `fd` {number|undefined} The file descriptor for operations that operate on
+  an existing file descriptor, such as `read`, `write`, `fsync`, or `close`.
+
+Large read/write buffers are not copied into the event payload. The `start`
+and `asyncStart` events carry no `result` or `error`; the `end` and `asyncEnd`
+events carry the `result` of the operation, and the `error` event carries the
+`error`, following the [TracingChannel Channels][] conventions.
+
+Operations performed through streams (`fs.createReadStream` and
+`fs.createWriteStream`) and most `FileHandle` methods are not covered by these
+channel families, and may not emit the full set of events.
+
+##### Event: `'tracing:fs.<operation>:start'`
+
+Emitted synchronously when an operation begins, before the operation is
+submitted. For synchronous operations this is followed by `end` (or `error`);
+for asynchronous operations it is followed by `end` and then `asyncStart`/
+`asyncEnd` (or `error`).
+
+##### Event: `'tracing:fs.<operation>:end'`
+
+* `result` {any} The result of the operation.
+
+Emitted when the operation completes. For synchronous operations this carries
+the operation `result`; for asynchronous operations it is emitted when the
+operation is submitted and carries no `result` (the `result` is delivered on
+the `asyncEnd` event).
+
+##### Event: `'tracing:fs.<operation>:asyncStart'`
+
+Emitted when the asynchronous work for an operation begins (when the
+completion callback is invoked).
+
+##### Event: `'tracing:fs.<operation>:asyncEnd'`
+
+* `result` {any} The result of the operation.
+
+Emitted when the asynchronous work for an operation completes, carrying the
+operation `result`.
+
+##### Event: `'tracing:fs.<operation>:error'`
+
+* `error` {Error} The error that caused the operation to fail.
+
+Emitted when an operation fails.
+
 #### HTTP
 
 > Stability: 1 - Experimental

src/env_properties.h

@@ -87,6 +87,7 @@
   V(allow_bare_named_params_string, "allowBareNamedParameters")                \
   V(allow_unknown_named_params_string, "allowUnknownNamedParameters")          \
   V(alpn_callback_string, "ALPNCallback")                                      \
+  V(api_string, "api")                                                         \
   V(args_string, "args")                                                       \
   V(arguments_string, "arguments")                                             \
   V(async_ids_stack_string, "async_ids_stack")                                 \

src/node_file-inl.h

@@ -220,6 +220,7 @@ FSReqPromise<AliasedBufferT>::FSReqPromise(BindingData* binding_data,
 template <typename AliasedBufferT>
 void FSReqPromise<AliasedBufferT>::Reject(v8::Local<v8::Value> reject) {
   finished_ = true;
+  PublishFSOpCompletionEvent(this, FSOperationChannel::kError, "error", reject);
   v8::HandleScope scope(env()->isolate());
   InternalCallbackScope callback_scope(this);
   v8::Local<v8::Value> value;
@@ -238,6 +239,8 @@ void FSReqPromise<AliasedBufferT>::Reject(v8::Local<v8::Value> reject) {
 template <typename AliasedBufferT>
 void FSReqPromise<AliasedBufferT>::Resolve(v8::Local<v8::Value> value) {
   finished_ = true;
+  PublishFSOpCompletionEvent(this, FSOperationChannel::kAsyncEnd, "result",
+                             value);
   v8::HandleScope scope(env()->isolate());
   InternalCallbackScope callback_scope(this);
   v8::Local<v8::Value> val;
@@ -303,6 +306,7 @@ FSReqBase* GetReqWrap(const v8::FunctionCallbackInfo<v8::Value>& args,
         result =
             FSReqPromise<AliasedFloat64Array>::New(binding_data, use_bigint);
       }
+      result->set_is_promise(true);
     }
   }
   if (result != nullptr) {
@@ -320,13 +324,32 @@ FSReqBase* AsyncDestCall(Environment* env, FSReqBase* req_wrap,
                          Func fn, Args... fn_args) {
   CHECK_NOT_NULL(req_wrap);
   req_wrap->Init(syscall, dest, len, enc);
+  BindingData* binding = req_wrap->binding_data();
+  const char* api = req_wrap->is_promise() ? "promise" : "callback";
+  std::string dest_str;
+  if (binding != nullptr) {
+    if (req_wrap->data() != nullptr) dest_str = req_wrap->data();
+    PublishFSOperationEvent(binding, env, FSOperationChannel::kStart, syscall,
+                            api, std::string(), dest_str, -1, nullptr,
+                            v8::Local<v8::Value>());
+  }
   int err = req_wrap->Dispatch(fn, fn_args..., after);
   if (err < 0) {
     uv_fs_t* uv_req = req_wrap->req();
     uv_req->result = err;
     uv_req->path = nullptr;
     after(uv_req);  // after may delete req_wrap if there is an error
     req_wrap = nullptr;
+  } else if (binding != nullptr) {
+    std::string path;
+    if (req_wrap->req()->path != nullptr) path = req_wrap->req()->path;
+    int fd = -1;
+    if (OperationUsesFd(req_wrap->req()->fs_type)) fd = req_wrap->req()->file;
+    req_wrap->set_op_path(path);
+    req_wrap->set_fd(fd);
+    PublishFSOperationEvent(binding, env, FSOperationChannel::kEnd, syscall,
+                            api, path, dest_str, fd, nullptr,
+                            v8::Local<v8::Value>());
   }
   return req_wrap;
 }
@@ -381,7 +404,43 @@ int SyncCallAndThrowIf(Predicate should_throw,
                        Func fn,
                        Args... args) {
   env->PrintSyncTrace();
+  BindingData* binding = Realm::GetBindingData<BindingData>(env->context());
+  std::string path;
+  std::string dest;
+  if (binding != nullptr) {
+    if (req_wrap->path_p != nullptr) path = req_wrap->path_p;
+    if (req_wrap->dest_p != nullptr) dest = req_wrap->dest_p;
+    PublishFSOperationEvent(binding, env, FSOperationChannel::kStart,
+                            req_wrap->syscall_p, "sync", path, dest, -1,
+                            nullptr, v8::Local<v8::Value>());
+  }
   int result = fn(nullptr, &(req_wrap->req), args..., nullptr);
+  if (binding != nullptr) {
+    int fd = -1;
+    if (OperationUsesFd(req_wrap->req.fs_type)) fd = req_wrap->req.file;
+    if (should_throw(result)) {
+      v8::Local<v8::Value> error = UVException(env->isolate(),
+                                               result,
+                                               req_wrap->syscall_p,
+                                               nullptr,
+                                               req_wrap->path_p,
+                                               req_wrap->dest_p);
+      PublishFSOperationEvent(binding, env, FSOperationChannel::kError,
+                              req_wrap->syscall_p, "sync", path, dest, fd,
+                              "error", error);
+    } else {
+      PublishFSOperationEvent(binding,
+                              env,
+                              FSOperationChannel::kEnd,
+                              req_wrap->syscall_p,
+                              "sync",
+                              path,
+                              dest,
+                              fd,
+                              "result",
+                              v8::Integer::New(env->isolate(), result));
+    }
+  }
   if (should_throw(result)) {
     env->ThrowUVException(result,
                           req_wrap->syscall_p,

src/node_file.cc

@@ -87,6 +87,120 @@ using v8::TryCatch;
 using v8::Undefined;
 using v8::Value;
 
+// Event names for the built-in per-operation fs tracing channel families,
+// one per FSOperationChannel in node_file.h. Each operation gets its own
+// channel family named `tracing:fs.<operation>:<event>`.
+const char* const kFSOperationEventNames[kNumFSOperationChannels] = {
+    "start",
+    "end",
+    "asyncStart",
+    "asyncEnd",
+    "error",
+};
+
+void PublishFSOperationEvent(BindingData* binding,
+                             Environment* env,
+                             FSOperationChannel channel,
+                             const char* operation,
+                             const char* api,
+                             const std::string& path,
+                             const std::string& dest,
+                             int fd,
+                             const char* value_key,
+                             Local<Value> value) {
+  const size_t index = static_cast<size_t>(channel);
+  CHECK_LT(index, kNumFSOperationChannels);
+  auto& channels = binding->fs_op_channels_[operation];
+  diagnostics_channel::Channel* ch = channels[index];
+  if (ch == nullptr) {
+    std::string name = std::string("tracing:fs.") + operation + ":" +
+                       kFSOperationEventNames[index];
+    ch = diagnostics_channel::Channel::Get(env, name.c_str());
+    channels[index] = ch;
+  }
+  if (ch == nullptr || !ch->HasSubscribers()) {
+    return;
+  }
+
+  Isolate* isolate = env->isolate();
+  HandleScope scope(isolate);
+  Local<Context> context = env->context();
+  Local<Object> obj = Object::New(isolate);
+  obj->Set(context, env->api_string(),
+           String::NewFromUtf8(isolate, api).ToLocalChecked())
+      .Check();
+  if (!path.empty()) {
+    obj->Set(context,
+             env->path_string(),
+             String::NewFromUtf8(isolate,
+                                 path.data(),
+                                 v8::NewStringType::kNormal,
+                                 static_cast<int>(path.size()))
+                 .ToLocalChecked())
+        .Check();
+  }
+  if (!dest.empty()) {
+    obj->Set(context,
+             env->dest_string(),
+             String::NewFromUtf8(isolate,
+                                 dest.data(),
+                                 v8::NewStringType::kNormal,
+                                 static_cast<int>(dest.size()))
+                 .ToLocalChecked())
+        .Check();
+  }
+  if (fd != -1) {
+    obj->Set(context, env->fd_string(), Integer::New(isolate, fd)).Check();
+  }
+  if (value_key != nullptr && !value.IsEmpty()) {
+    obj->Set(context, OneByteString(isolate, value_key), value).Check();
+  }
+  ch->Publish(env, obj);
+}
+
+void PublishFSOpCompletionEvent(FSReqBase* req_wrap,
+                                FSOperationChannel channel,
+                                const char* value_key,
+                                Local<Value> value) {
+  BindingData* binding = req_wrap->binding_data();
+  if (binding == nullptr) return;
+  const char* api = req_wrap->is_promise() ? "promise" : "callback";
+  std::string dest;
+  if (req_wrap->data() != nullptr) dest = req_wrap->data();
+  PublishFSOperationEvent(binding,
+                          req_wrap->env(),
+                          channel,
+                          req_wrap->syscall(),
+                          api,
+                          req_wrap->op_path(),
+                          dest,
+                          req_wrap->fd(),
+                          value_key,
+                          value);
+}
+
+// Returns true if the libuv fs request type operates on an existing file
+// descriptor (as opposed to taking a path). These are the request types whose
+// `file` field holds the input descriptor.
+bool OperationUsesFd(uv_fs_type fs_type) {
+  switch (fs_type) {
+    case UV_FS_CLOSE:
+    case UV_FS_READ:
+    case UV_FS_WRITE:
+    case UV_FS_FSTAT:
+    case UV_FS_FTRUNCATE:
+    case UV_FS_FDATASYNC:
+    case UV_FS_FSYNC:
+    case UV_FS_FUTIME:
+    case UV_FS_FCHMOD:
+    case UV_FS_FCHOWN:
+    case UV_FS_SENDFILE:
+      return true;
+    default:
+      return false;
+  }
+}
+
 #ifndef S_ISDIR
 #define S_ISDIR(mode) (((mode)&S_IFMT) == S_IFDIR)
 #endif
@@ -221,6 +335,7 @@ FSReqBase::~FSReqBase() = default;
 
 void FSReqBase::MemoryInfo(MemoryTracker* tracker) const {
   tracker->TrackField("continuation_data", continuation_data_);
+  tracker->TrackField("op_path", op_path_);
 }
 
 // The FileHandle object wraps a file descriptor and will close it on garbage
@@ -728,6 +843,7 @@ int FileHandle::DoShutdown(ShutdownWrap* req_wrap) {
 }
 
 void FSReqCallback::Reject(Local<Value> reject) {
+  PublishFSOpCompletionEvent(this, FSOperationChannel::kError, "error", reject);
   MakeCallback(env()->oncomplete_string(), 1, &reject);
 }
 
@@ -740,6 +856,8 @@ void FSReqCallback::ResolveStatFs(const uv_statfs_t* stat) {
 }
 
 void FSReqCallback::Resolve(Local<Value> value) {
+  PublishFSOpCompletionEvent(this, FSOperationChannel::kAsyncEnd, "result",
+                             value);
   Local<Value> argv[2]{Null(env()->isolate()), value};
   MakeCallback(env()->oncomplete_string(),
                value->IsUndefined() ? 1 : arraysize(argv),
@@ -762,6 +880,10 @@ FSReqAfterScope::FSReqAfterScope(FSReqBase* wrap, uv_fs_t* req)
       handle_scope_(wrap->env()->isolate()),
       context_scope_(wrap->env()->context()) {
   CHECK_EQ(wrap_->req(), req);
+  // The async work for the operation has completed; the continuation window
+  // begins here.
+  PublishFSOpCompletionEvent(wrap, FSOperationChannel::kAsyncStart, nullptr,
+                             Local<Value>());
 }
 
 FSReqAfterScope::~FSReqAfterScope() {