Fixes

Modules/app-module/include/app/file.h

@@ -33,6 +33,12 @@ struct AppFileOps {
     error_t (*await)(void* object, AppFileWait wait, TickType_t timeout);
     /** @return bitmask of APP_FILE_READABLE / APP_FILE_WRITABLE. */
     uint32_t (*poll)(void* object);
+    /** Optional (may be NULL). Called by app_fd_table_get_and_retain() atomically with the fd
+     * lookup, before handing `object` to a caller about to dispatch into it, so a concurrent
+     * teardown of `object` can wait for that dispatch to finish instead of racing it. The caller
+     * calls release() once done, regardless of what the dispatched call returned. */
+    void (*retain)(void* object);
+    void (*release)(void* object);
 };
 
 /** A file-descriptor-table entry: an operations table paired with the object it operates on. */

Modules/app-module/private/app/private/fd_table.h

@@ -56,6 +56,15 @@ error_t app_fd_table_allocate(struct AppFdTable* table, const struct AppFileOps*
 /** @return true and fills @a out_file if @a fd is currently in use, false otherwise. */
 bool app_fd_table_get(struct AppFdTable* table, int fd, struct AppFile* out_file);
 
+/**
+ * Like app_fd_table_get(), but also calls the returned AppFile's AppFileOps::retain() (if any)
+ * while still holding `mutex`, atomically with the lookup. A caller that gets true back can
+ * therefore never have @a fd closed/torn down out from under it between this call and actually
+ * dispatching into the returned AppFile. Caller must call AppFileOps::release() once done
+ * dispatching.
+ */
+bool app_fd_table_get_and_retain(struct AppFdTable* table, int fd, struct AppFile* out_file);
+
 /**
  * Closes @a fd (via AppFileOps::close()) and frees the slot for reuse.
  * @retval ERROR_NOT_FOUND @a fd is out of range or not currently in use

Modules/app-module/source/fd_table.cpp

@@ -12,6 +12,24 @@ bool fd_in_range(int fd) {
     return fd >= 0 && fd < APP_MAX_FDS;
 }
 
+bool get_internal(AppFdTable* table, int fd, AppFile* out_file, bool retain) {
+    if (!fd_in_range(fd)) {
+        return false;
+    }
+
+    mutex_lock(&table->mutex);
+    check(!table->shutting_down);
+    bool in_use = table->fds[fd] != nullptr;
+    if (in_use) {
+        *out_file = table->slots[fd];
+        if (retain && out_file->ops->retain != nullptr) {
+            out_file->ops->retain(out_file->object);
+        }
+    }
+    mutex_unlock(&table->mutex);
+    return in_use;
+}
+
 } // namespace
 
 extern "C" {
@@ -86,18 +104,11 @@ error_t app_fd_table_allocate(AppFdTable* table, const AppFileOps* ops, void* ob
 }
 
 bool app_fd_table_get(AppFdTable* table, int fd, AppFile* out_file) {
-    if (!fd_in_range(fd)) {
-        return false;
-    }
+    return get_internal(table, fd, out_file, /*retain=*/false);
+}
 
-    mutex_lock(&table->mutex);
-    check(!table->shutting_down);
-    bool in_use = table->fds[fd] != nullptr;
-    if (in_use) {
-        *out_file = table->slots[fd];
-    }
-    mutex_unlock(&table->mutex);
-    return in_use;
+bool app_fd_table_get_and_retain(AppFdTable* table, int fd, AppFile* out_file) {
+    return get_internal(table, fd, out_file, /*retain=*/true);
 }
 
 error_t app_fd_table_close(AppFdTable* table, int fd) {

Modules/app-module/source/io.cpp

@@ -49,11 +49,15 @@ ssize_t app_io_read(int fd, void* buffer, size_t size) {
     }
 
     AppFile file {};
-    if (!app_fd_table_get(table, fd, &file)) {
+    if (!app_fd_table_get_and_retain(table, fd, &file)) {
         errno = EBADF;
         return -1;
     }
-    return file.ops->read(file.object, buffer, size);
+    ssize_t result = file.ops->read(file.object, buffer, size);
+    if (file.ops->release != nullptr) {
+        file.ops->release(file.object);
+    }
+    return result;
 }
 
 ssize_t app_io_write(int fd, const void* buffer, size_t size) {
@@ -67,11 +71,15 @@ ssize_t app_io_write(int fd, const void* buffer, size_t size) {
     }
 
     AppFile file {};
-    if (!app_fd_table_get(table, fd, &file)) {
+    if (!app_fd_table_get_and_retain(table, fd, &file)) {
         errno = EBADF;
         return -1;
     }
-    return file.ops->write(file.object, buffer, size);
+    ssize_t result = file.ops->write(file.object, buffer, size);
+    if (file.ops->release != nullptr) {
+        file.ops->release(file.object);
+    }
+    return result;
 }
 
 int app_io_close(int fd) {

Modules/app-module/source/manager.cpp

@@ -126,6 +126,12 @@ error_t start_internal(const char* id, AppInstanceId parent_instance_id, int arg
         error_t bind_result = app_stream_subscribe(bindings[i].stream, bindings[i].buffer, bindings[i].buffer_capacity, bindings[i].event_group, target_id, bindings[i].producer_fd);
         if (bind_result != ERROR_NONE) {
             LOG_E(TAG, "[instance %d] Failed to bind stream at fd %d: %s", target_id, bindings[i].producer_fd, error_to_string(bind_result));
+            // Undo bindings[0..i): app_fd_table_teardown() below only closes each stream. It
+            // doesn't release the event bits app_stream_subscribe() claimed or destruct
+            // stream->internal.mutex; only app_stream_unsubscribe() does that.
+            for (size_t j = 0; j < i; j++) {
+                app_stream_unsubscribe(bindings[j].stream);
+            }
             mutex_lock(&ledger.mutex);
             app_fd_table_teardown(&ledger.instances[target_id].fd_table);
             ledger.instances.erase(target_id);
@@ -137,6 +143,11 @@ error_t start_internal(const char* id, AppInstanceId parent_instance_id, int arg
 
     error_t error = app_scheduler_start(target_id, manifest->location, manifest->stack, argc, argv);
     if (error != ERROR_NONE) {
+        // Every binding succeeded before app_scheduler_start() failed. Unsubscribe all of them,
+        // same reasoning as the bind-failure path above.
+        for (size_t j = 0; j < binding_count; j++) {
+            app_stream_unsubscribe(bindings[j].stream);
+        }
         mutex_lock(&ledger.mutex);
         app_fd_table_teardown(&ledger.instances[target_id].fd_table);
         ledger.instances.erase(target_id);

Modules/app-module/source/stream.cpp

@@ -113,12 +113,31 @@ uint32_t stream_file_poll(void* object) {
     return bits;
 }
 
+// Fused with app_fd_table_get_and_retain()'s own lock, so a caller that gets a live AppFile back
+// is already counted here before it can be paused and raced by app_stream_unsubscribe()'s drain
+// wait (see StreamOperationGuard above for the equivalent per-call bracketing).
+void stream_file_retain(void* object) {
+    auto* stream = static_cast<AppStream*>(object);
+    mutex_lock(&stream->internal.mutex);
+    stream->internal.active_operations++;
+    mutex_unlock(&stream->internal.mutex);
+}
+
+void stream_file_release(void* object) {
+    auto* stream = static_cast<AppStream*>(object);
+    mutex_lock(&stream->internal.mutex);
+    stream->internal.active_operations--;
+    mutex_unlock(&stream->internal.mutex);
+}
+
 constexpr AppFileOps STREAM_OPS = {
     .read = stream_file_read,
     .write = stream_file_write,
     .close = stream_file_close,
     .await = stream_file_await,
     .poll = stream_file_poll,
+    .retain = stream_file_retain,
+    .release = stream_file_release,
 };
 
 } // namespace

Modules/app-module/tests/source/io_test.cpp

@@ -170,3 +170,39 @@ TEST_CASE("a write blocked on a full stream wakes with an error once the consume
     task_event_group_destruct(&event_group);
     app_manager_remove("test.io.blocked");
 }
+
+TEST_CASE("app_stream_unsubscribe is safe to call while a write is actively blocked") {
+    ensure_memory_loader_registered();
+    g_blocked_writer_saw_error.store(false, std::memory_order_relaxed);
+    g_blocked_writer_done.store(false, std::memory_order_relaxed);
+
+    AppManifest manifest { "test.io.unsub_race", "UnsubRace", APP_CATEGORY_USER, { APP_LOCATION_MEMORY, reinterpret_cast<void*>(blocked_writer_app_main) } };
+    REQUIRE_EQ(app_manager_add(&manifest), ERROR_NONE);
+
+    TaskEventGroup event_group {};
+    task_event_group_construct(&event_group);
+
+    uint8_t storage[4]; // smaller than the 10 bytes blocked_writer_app_main sends
+    AppStream child_stdout {};
+
+    AppStreamBinding binding { STDOUT_FILENO, &child_stdout, storage, sizeof(storage), &event_group };
+    AppInstanceId child_id = 0;
+    REQUIRE_EQ(app_manager_start_with_streams("test.io.unsub_race", &binding, 1, &child_id), ERROR_NONE);
+
+    // Give the child time to fill the 4-byte buffer and block inside app_io_write(), already
+    // dispatched through app_fd_table_get_and_retain() and currently waiting in
+    // app_stream_await(). This is the exact state app_stream_unsubscribe() must be safe to run
+    // against, with no prior app_stream_close() or wait for the child to stop first.
+    delay_millis(200);
+    REQUIRE_FALSE(g_blocked_writer_done.load(std::memory_order_acquire));
+
+    // Regression: unsubscribing directly here used to be able to destruct
+    // stream->internal.mutex while the blocked write above was still executing against it.
+    REQUIRE_EQ(app_stream_unsubscribe(&child_stdout), ERROR_NONE);
+
+    REQUIRE(wait_for_state(child_id, APP_INSTANCE_STATE_STOPPED, 1000));
+    CHECK(g_blocked_writer_saw_error.load(std::memory_order_acquire));
+
+    task_event_group_destruct(&event_group);
+    app_manager_remove("test.io.unsub_race");
+}