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

@@ -1636,6 +1636,89 @@ diagnosticsChannel.subscribe('crypto.fips.indicator', (message) => {
 });
 ```
 
+#### 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`), most `FileHandle` methods, and the `fs.readFile`
+fast path (which batches open/stat/read/close into a single background job)
+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

@@ -214,6 +214,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;
@@ -232,6 +233,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;
@@ -311,6 +314,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) {
@@ -328,13 +332,40 @@ 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";
+  FSOperationChannels* channels = nullptr;
+  // See SyncCallAndThrowIf: instrumentation is unsafe with a pending
+  // exception.
+  if (binding != nullptr && !env->isolate()->HasPendingException()) {
+    channels = &GetFSOperationChannels(binding, env, syscall);
+    req_wrap->set_op_channels(channels);
+    if (FSOperationChannelHasSubscribers(*channels,
+                                         FSOperationChannel::kStart)) {
+      PublishFSOperationEvent(env, *channels, FSOperationChannel::kStart,
+                              api, nullptr, req_wrap->data(), -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 (channels != nullptr &&
+             AnyFSOperationChannelHasSubscribers(*channels)) {
+    const char* path = req_wrap->req()->path;
+    int fd = -1;
+    if (OperationUsesFd(req_wrap->req()->fs_type)) fd = req_wrap->req()->file;
+    req_wrap->set_fd(fd);
+    // The path is captured for the completion events; it requires a copy
+    // since the uv request is cleaned up before they fire.
+    req_wrap->set_op_path(path == nullptr ? std::string() : path);
+    PublishFSOperationEvent(env, *channels, FSOperationChannel::kEnd, api,
+                            path, req_wrap->data(), fd, nullptr,
+                            v8::Local<v8::Value>());
   }
   return req_wrap;
 }
@@ -389,7 +420,54 @@ int SyncCallAndThrowIf(Predicate should_throw,
                        Func fn,
                        Args... args) {
   env->PrintSyncTrace();
+  BindingData* binding = Realm::GetBindingData<BindingData>(env->context());
+  FSOperationChannels* channels = nullptr;
+  // The instrumentation creates V8 objects and may run subscribers, neither
+  // of which is safe with a pending exception (a multi-step operation keeps
+  // going after a failed step to clean up, e.g. write + close).
+  if (binding != nullptr && !env->isolate()->HasPendingException()) {
+    channels = &GetFSOperationChannels(binding, env, req_wrap->syscall_p);
+    if (FSOperationChannelHasSubscribers(*channels,
+                                         FSOperationChannel::kStart)) {
+      PublishFSOperationEvent(env, *channels, FSOperationChannel::kStart,
+                              "sync", req_wrap->path_p, req_wrap->dest_p, -1,
+                              nullptr, v8::Local<v8::Value>());
+    }
+  }
   int result = fn(nullptr, &(req_wrap->req), args..., nullptr);
+  if (channels != nullptr) {
+    if (should_throw(result)) {
+      // The error object is only built when someone is listening; the throw
+      // path below creates its own copy.
+      if (FSOperationChannelHasSubscribers(*channels,
+                                           FSOperationChannel::kError)) {
+        int fd = -1;
+        if (OperationUsesFd(req_wrap->req.fs_type)) fd = req_wrap->req.file;
+        v8::Local<v8::Value> error = UVException(env->isolate(),
+                                                 result,
+                                                 req_wrap->syscall_p,
+                                                 nullptr,
+                                                 req_wrap->path_p,
+                                                 req_wrap->dest_p);
+        PublishFSOperationEvent(env, *channels, FSOperationChannel::kError,
+                                "sync", req_wrap->path_p, req_wrap->dest_p,
+                                fd, "error", error);
+      }
+    } else if (FSOperationChannelHasSubscribers(*channels,
+                                                FSOperationChannel::kEnd)) {
+      int fd = -1;
+      if (OperationUsesFd(req_wrap->req.fs_type)) fd = req_wrap->req.file;
+      PublishFSOperationEvent(env,
+                              *channels,
+                              FSOperationChannel::kEnd,
+                              "sync",
+                              req_wrap->path_p,
+                              req_wrap->dest_p,
+                              fd,
+                              "result",
+                              v8::Integer::New(env->isolate(), result));
+    }
+  }
   if (should_throw(result)) {
     env->ThrowUVException(result,
                           req_wrap->syscall_p,

src/node_file.cc

@@ -93,6 +93,125 @@ using v8::Uint8Array;
 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",
+};
+
+FSOperationChannels& GetFSOperationChannels(BindingData* binding,
+                                            Environment* env,
+                                            const char* operation) {
+  auto& names = binding->fs_op_channel_names_;
+  for (size_t i = 0; i < names.size(); i++) {
+    if (names[i] == operation) {
+      return *binding->fs_op_channel_sets_[i];
+    }
+  }
+  auto set = std::make_unique<FSOperationChannels>();
+  for (size_t i = 0; i < kNumFSOperationChannels; i++) {
+    std::string name = std::string("tracing:fs.") + operation + ":" +
+                       kFSOperationEventNames[i];
+    (*set)[i] = diagnostics_channel::Channel::Get(env, name);
+  }
+  names.push_back(operation);
+  binding->fs_op_channel_sets_.push_back(std::move(set));
+  return *binding->fs_op_channel_sets_.back();
+}
+
+void PublishFSOperationEvent(Environment* env,
+                             FSOperationChannels& channels,
+                             FSOperationChannel channel,
+                             const char* api,
+                             const char* path,
+                             const char* dest,
+                             int fd,
+                             const char* value_key,
+                             Local<Value> value) {
+  const size_t index = static_cast<size_t>(channel);
+  CHECK_LT(index, kNumFSOperationChannels);
+  diagnostics_channel::Channel* ch = channels[index].get();
+  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(),
+           ToV8Value(context, api, isolate).ToLocalChecked())
+      .Check();
+  if (path != nullptr && path[0] != '\0') {
+    obj->Set(context,
+             env->path_string(),
+             ToV8Value(context, path, isolate).ToLocalChecked())
+        .Check();
+  }
+  if (dest != nullptr && dest[0] != '\0') {
+    obj->Set(context,
+             env->dest_string(),
+             ToV8Value(context, dest, isolate).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) {
+  FSOperationChannels* channels = req_wrap->op_channels();
+  if (channels == nullptr ||
+      !FSOperationChannelHasSubscribers(*channels, channel)) {
+    return;
+  }
+  const char* api = req_wrap->is_promise() ? "promise" : "callback";
+  PublishFSOperationEvent(req_wrap->env(),
+                          *channels,
+                          channel,
+                          api,
+                          req_wrap->op_path().c_str(),
+                          req_wrap->data(),
+                          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
@@ -227,6 +346,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
@@ -734,6 +854,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);
 }
 
@@ -746,6 +867,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),
@@ -774,6 +897,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() {