Merge pull request #14 from kimlherme/fix/audio-jni-ownership

fix(runtime): preserve OpenSL tickets and JNI array references

src/audio/opensl_simple_buffer_queue.cc

@@ -12,9 +12,7 @@
 #include <mutex>
 #include <new>
 #include <string>
-#include <unordered_map>
 #include <utility>
-#include <vector>
 
 namespace mocktail::audio {
 namespace {
@@ -55,10 +53,8 @@ struct OpenSlSimpleBufferQueueAdapter::State
   };
 
   struct ReleaseTicket {
-    std::shared_ptr<State> state;
-    std::uint64_t id = 0;
-    std::uint64_t generation = 0;
-    bool in_use = false;
+    const std::shared_ptr<State> state;
+    const std::uint64_t generation;
   };
 
   struct CallbackJob {
@@ -71,13 +67,7 @@ struct OpenSlSimpleBufferQueueAdapter::State
       : sink(std::move(owned_sink)),
         max_buffers(options.max_buffers),
         event_callback(options.event_callback),
-        event_context(options.event_context),
-        ticket_pool(options.max_buffers) {
-    free_tickets.reserve(options.max_buffers);
-    for (std::size_t i = options.max_buffers; i > 0; --i) {
-      free_tickets.push_back(i - 1);
-    }
-  }
+        event_context(options.event_context) {}
 
   static const AndroidSimpleBufferQueueTable kQueueTable;
 
@@ -103,7 +93,6 @@ struct OpenSlSimpleBufferQueueAdapter::State
     const std::shared_ptr<State> state = raw_state->shared_from_this();
 
     std::lock_guard<std::mutex> operation_lock(state->sink_operation_mutex);
-    std::uint64_t id = 0;
     ReleaseTicket* ticket = nullptr;
     {
       std::lock_guard<std::mutex> lock(state->mutex);
@@ -111,24 +100,24 @@ struct OpenSlSimpleBufferQueueAdapter::State
         return opensl_abi::kResultPreconditionsViolated;
       }
       if (state->pending_count.load(std::memory_order_relaxed) >=
-              state->max_buffers ||
-          state->free_tickets.empty()) {
+          state->max_buffers) {
         return opensl_abi::kResultBufferInsufficient;
       }
-      const std::size_t slot_index = state->free_tickets.back();
-      state->free_tickets.pop_back();
-      ticket = &state->ticket_pool[slot_index];
-      id = state->next_id++;
-      ticket->state = state;
-      ticket->id = id;
-      ticket->generation = state->generation;
-      ticket->in_use = true;
+      ticket = new (std::nothrow)
+          ReleaseTicket{state, state->generation};
+      if (ticket == nullptr) {
+        state->last_status = Status::Error(
+            StatusCode::kUnavailable, "unable to allocate OpenSL release ticket");
+        return ToOpenSlResult(state->last_status);
+      }
       state->pending_count.fetch_add(1, std::memory_order_relaxed);
       ++state->submitted_buffers;
     }
 
     const PcmBuffer pcm{buffer, static_cast<std::size_t>(size),
                         &State::ReleaseBuffer, ticket};
+    // A successful enqueue may release synchronously. Only a failed enqueue
+    // leaves ownership here; its contract forbids a release callback.
     const Status status = state->sink->Enqueue(pcm);
     if (status.ok()) {
       state->NotifyEvent(OpenSlBufferQueueEvent::kSubmitted,
@@ -136,22 +125,12 @@ struct OpenSlSimpleBufferQueueAdapter::State
       return opensl_abi::kResultSuccess;
     }
 
+    const std::unique_ptr<ReleaseTicket> failed_ticket(ticket);
     {
       std::lock_guard<std::mutex> lock(state->mutex);
       state->last_status = status;
-      if (ticket->in_use && ticket->id == id) {
-        ticket->in_use = false;
-        ticket->state.reset();
-        const std::size_t slot_index =
-            static_cast<std::size_t>(ticket - state->ticket_pool.data());
-        state->free_tickets.push_back(slot_index);
-        if (state->pending_count.load(std::memory_order_relaxed) > 0) {
-          state->pending_count.fetch_sub(1, std::memory_order_relaxed);
-        }
-        if (state->submitted_buffers > 0) {
-          --state->submitted_buffers;
-        }
-      }
+      state->pending_count.fetch_sub(1, std::memory_order_relaxed);
+      --state->submitted_buffers;
     }
     return ToOpenSlResult(status);
   }
@@ -164,25 +143,14 @@ struct OpenSlSimpleBufferQueueAdapter::State
     }
     const std::shared_ptr<State> state = raw_state->shared_from_this();
     std::lock_guard<std::mutex> operation_lock(state->sink_operation_mutex);
-    std::size_t discarded_count = 0;
     {
       std::lock_guard<std::mutex> lock(state->mutex);
       if (state->stopping) {
         return opensl_abi::kResultPreconditionsViolated;
       }
       ++state->generation;
-      state->free_tickets.clear();
-      for (auto& ticket : state->ticket_pool) {
-        if (ticket.in_use) {
-          ticket.in_use = false;
-          ticket.state.reset();
-          ++discarded_count;
-        }
-      }
-      for (std::size_t i = state->max_buffers; i > 0; --i) {
-        state->free_tickets.push_back(i - 1);
-      }
-      state->discarded_buffers += discarded_count;
+      state->discarded_buffers +=
+          state->pending_count.load(std::memory_order_relaxed);
       state->pending_count.store(0, std::memory_order_relaxed);
     }
     const Status status = state->sink->Clear();
@@ -226,7 +194,8 @@ struct OpenSlSimpleBufferQueueAdapter::State
 
   static void ReleaseBuffer(void* context, const void* /*data*/,
                             std::size_t size_bytes) {
-    auto* ticket = static_cast<ReleaseTicket*>(context);
+    const std::unique_ptr<ReleaseTicket> ticket(
+        static_cast<ReleaseTicket*>(context));
     if (ticket == nullptr) {
       return;
     }
@@ -239,29 +208,27 @@ struct OpenSlSimpleBufferQueueAdapter::State
     bool discarded = false;
     {
       std::lock_guard<std::mutex> lock(state->mutex);
-      if (ticket->in_use) {
-        ticket->in_use = false;
-        ticket->state.reset();
-        const std::size_t slot_index =
-            static_cast<std::size_t>(ticket - state->ticket_pool.data());
-        state->free_tickets.push_back(slot_index);
-        if (state->pending_count.load(std::memory_order_relaxed) > 0) {
-          state->pending_count.fetch_sub(1, std::memory_order_relaxed);
-        }
-        if (!state->stopping && ticket->generation == state->generation) {
-          state->processed_buffers.fetch_add(1, std::memory_order_relaxed);
-          ++state->consumed_buffers;
-          state->consumed_bytes += size_bytes;
-          consumed = true;
-          if (state->callback != nullptr) {
-            state->callback_jobs.push_back(
-                CallbackJob{state->callback, state->callback_context});
-            notify_worker = true;
-          }
-        } else {
-          ++state->discarded_buffers;
-          discarded = true;
+      // Clear already counted older generations as discarded. Their tickets
+      // remain alive until release without affecting newly enqueued buffers.
+      if (ticket->generation != state->generation) {
+        return;
+      }
+      if (state->pending_count.load(std::memory_order_relaxed) > 0) {
+        state->pending_count.fetch_sub(1, std::memory_order_relaxed);
+      }
+      if (!state->stopping) {
+        state->processed_buffers.fetch_add(1, std::memory_order_relaxed);
+        ++state->consumed_buffers;
+        state->consumed_bytes += size_bytes;
+        consumed = true;
+        if (state->callback != nullptr) {
+          state->callback_jobs.push_back(
+              CallbackJob{state->callback, state->callback_context});
+          notify_worker = true;
         }
+      } else {
+        ++state->discarded_buffers;
+        discarded = true;
       }
     }
     if (consumed) {
@@ -374,14 +341,11 @@ struct OpenSlSimpleBufferQueueAdapter::State
   void* const event_context;
   Handle handle;
   pthread_t callback_thread{};
-  std::vector<ReleaseTicket> ticket_pool;
-  std::vector<std::size_t> free_tickets;
   std::atomic<std::size_t> pending_count{0};
   std::deque<CallbackJob> callback_jobs;
   AndroidSimpleBufferQueueCallback callback = nullptr;
   void* callback_context = nullptr;
   Status last_status;
-  std::uint64_t next_id = 1;
   std::uint64_t generation = 1;
   std::atomic<std::uint64_t> processed_buffers{0};
   std::uint64_t submitted_buffers = 0;
@@ -547,7 +511,6 @@ void OpenSlSimpleBufferQueueAdapter::Shutdown() {
     }
     state->shutdown_started = true;
     state->stopping = true;
-    ++state->generation;
     state->callback = nullptr;
     state->callback_context = nullptr;
     state->callback_jobs.clear();
@@ -566,10 +529,6 @@ void OpenSlSimpleBufferQueueAdapter::Shutdown() {
       state->last_status = Status::Error(
           StatusCode::kPlatformError,
           "audio sink did not release all borrowed OpenSL buffers");
-      for (auto& ticket : state->ticket_pool) {
-        ticket.in_use = false;
-        ticket.state.reset();
-      }
       state->pending_count.store(0, std::memory_order_relaxed);
     }
     state->shutdown_complete = true;

src/jnivm/jnivm.cc

@@ -818,7 +818,13 @@ void ReleaseJniReference(jobject obj) {
   auto arr_it = g_array_storage.find(reinterpret_cast<jarray>(obj));
   if (arr_it != g_array_storage.end()) {
     std::unique_ptr<PseudoArray> dying_array = std::move(arr_it->second);
+    std::vector<jobject> elements = std::move(dying_array->objects);
     g_array_storage.erase(arr_it);
+    // Release after detaching the array: elements may recursively erase other
+    // arrays. Keep dying_array alive until those releases have completed.
+    for (jobject element : elements) {
+      ReleaseJniReference(element);
+    }
     return;
   }
   if (g_known_classes.find(reinterpret_cast<jclass>(obj)) !=
@@ -3929,6 +3935,11 @@ jobjectArray MakeObjectArray(jsize len, jobject init) {
   auto array = std::make_unique<PseudoArray>();
   if (len > 0) {
     array->objects.resize(static_cast<std::size_t>(len), init);
+    if (init != nullptr) {
+      for (jsize index = 0; index < len; ++index) {
+        RetainJniReference(init);
+      }
+    }
   }
   jobjectArray ref = reinterpret_cast<jobjectArray>(array.get());
   g_array_storage[ref] = std::move(array);

tests/audio_foundation_test.cc

@@ -30,19 +30,60 @@ namespace {
 
 using namespace std::chrono_literals;
 
+class ReleaseGate {
+ public:
+  void WaitBeforeRelease() {
+    std::unique_lock<std::mutex> lock(mutex_);
+    waiting_ = true;
+    cv_.notify_all();
+    cv_.wait(lock, [this] { return open_; });
+  }
+
+  bool WaitUntilBlocked() {
+    std::unique_lock<std::mutex> lock(mutex_);
+    return cv_.wait_for(lock, 2s, [this] { return waiting_; });
+  }
+
+  void Open() {
+    {
+      std::lock_guard<std::mutex> lock(mutex_);
+      open_ = true;
+    }
+    cv_.notify_all();
+  }
+
+ private:
+  std::mutex mutex_;
+  std::condition_variable cv_;
+  bool waiting_ = false;
+  bool open_ = false;
+};
+
 class ControlledAudioSink final : public AudioSink {
  public:
+  explicit ControlledAudioSink(bool release_on_enqueue = false)
+      : release_on_enqueue_(release_on_enqueue) {}
+
   const PcmSpec& source_spec() const override { return spec_; }
 
   Status Enqueue(const PcmBuffer& buffer) override {
     if (buffer.data == nullptr || buffer.size_bytes == 0) {
       return Status::Error(StatusCode::kInvalidArgument, "empty buffer");
     }
-    std::lock_guard<std::mutex> lock(mutex_);
-    if (stopped_) {
-      return Status::Error(StatusCode::kFailedPrecondition, "stopped");
+    {
+      std::lock_guard<std::mutex> lock(mutex_);
+      if (stopped_) {
+        return Status::Error(StatusCode::kFailedPrecondition, "stopped");
+      }
+      if (reject_next_enqueue_) {
+        reject_next_enqueue_ = false;
+        return Status::Error(StatusCode::kUnavailable, "enqueue rejected");
+      }
+      buffers_.push_back(buffer);
+    }
+    if (release_on_enqueue_) {
+      ReleaseOne();
     }
-    buffers_.push_back(buffer);
     return Status::Ok();
   }
 
@@ -93,10 +134,13 @@ class ControlledAudioSink final : public AudioSink {
       std::lock_guard<std::mutex> lock(mutex_);
       stopped_ = true;
     }
+    release_cv_.notify_all();
     ReleaseAll();
+    std::unique_lock<std::mutex> lock(mutex_);
+    release_cv_.wait(lock, [this] { return active_releases_ == 0; });
   }
 
-  void ReleaseOne() {
+  void ReleaseOne(ReleaseGate* gate = nullptr) {
     PcmBuffer buffer;
     {
       std::lock_guard<std::mutex> lock(mutex_);
@@ -105,11 +149,30 @@ class ControlledAudioSink final : public AudioSink {
       }
       buffer = buffers_.front();
       buffers_.erase(buffers_.begin());
+      ++active_releases_;
+    }
+    if (gate != nullptr) {
+      gate->WaitBeforeRelease();
     }
     if (buffer.release_callback != nullptr) {
       buffer.release_callback(buffer.release_context, buffer.data,
                               buffer.size_bytes);
     }
+    {
+      std::lock_guard<std::mutex> lock(mutex_);
+      --active_releases_;
+    }
+    release_cv_.notify_all();
+  }
+
+  bool WaitUntilShutdownStarts() {
+    std::unique_lock<std::mutex> lock(mutex_);
+    return release_cv_.wait_for(lock, 2s, [this] { return stopped_; });
+  }
+
+  void RejectNextEnqueue() {
+    std::lock_guard<std::mutex> lock(mutex_);
+    reject_next_enqueue_ = true;
   }
 
   std::size_t buffer_count() const {
@@ -141,11 +204,36 @@ class ControlledAudioSink final : public AudioSink {
   }
 
   const PcmSpec spec_;
+  const bool release_on_enqueue_;
   mutable std::mutex mutex_;
+  std::condition_variable release_cv_;
   std::vector<PcmBuffer> buffers_;
+  std::size_t active_releases_ = 0;
   bool playing_ = false;
   float gain_ = 1.0F;
   bool stopped_ = false;
+  bool reject_next_enqueue_ = false;
+};
+
+class DelayedBufferRelease {
+ public:
+  explicit DelayedBufferRelease(ControlledAudioSink* sink)
+      : thread_([this, sink] { sink->ReleaseOne(&gate_); }) {}
+
+  ~DelayedBufferRelease() {
+    if (thread_.joinable()) {
+      gate_.Open();
+      thread_.join();
+    }
+  }
+
+  bool WaitUntilBlocked() { return gate_.WaitUntilBlocked(); }
+  void AllowRelease() { gate_.Open(); }
+  void Join() { thread_.join(); }
+
+ private:
+  ReleaseGate gate_;
+  std::thread thread_;
 };
 
 struct CallbackProbe {
@@ -389,6 +477,7 @@ TEST(SdlAudioSinkTest, ReleasesBorrowedBufferWhenStreamIsCleared) {
 TEST(OpenSlQueueTest, DeliversPcmToSinkAndDispatchesCompletion) {
   auto controlled = std::make_unique<ControlledAudioSink>();
   ControlledAudioSink* controlled_view = controlled.get();
+  CallbackProbe probe;
   std::unique_ptr<OpenSlSimpleBufferQueueAdapter> adapter;
   Status status = OpenSlSimpleBufferQueueAdapter::Create(std::move(controlled),
                                                          {}, &adapter);
@@ -396,7 +485,6 @@ TEST(OpenSlQueueTest, DeliversPcmToSinkAndDispatchesCompletion) {
 
   auto queue = adapter->interface();
   ASSERT_NE(queue, nullptr);
-  CallbackProbe probe;
   ASSERT_EQ((*queue)->RegisterCallback(queue, OpenSlQueueCallback, &probe),
             opensl_abi::kResultSuccess);
 
@@ -520,6 +608,204 @@ TEST(OpenSlQueueTest, ClearSuppressesCallbacksForDiscardedBuffers) {
   EXPECT_EQ(probe.calls, 0);
 }
 
+TEST(OpenSlQueueTest, DelayedReleaseAfterClearPreservesReenqueuedBuffer) {
+  auto controlled = std::make_unique<ControlledAudioSink>();
+  ControlledAudioSink* controlled_view = controlled.get();
+  OpenSlSimpleBufferQueueOptions options;
+  options.max_buffers = 1;
+  CallbackProbe probe;
+  std::unique_ptr<OpenSlSimpleBufferQueueAdapter> adapter;
+  ASSERT_TRUE(OpenSlSimpleBufferQueueAdapter::Create(std::move(controlled),
+                                                     options, &adapter)
+                  .ok());
+  auto queue = adapter->interface();
+  ASSERT_EQ((*queue)->RegisterCallback(queue, OpenSlQueueCallback, &probe),
+            opensl_abi::kResultSuccess);
+  const std::vector<std::int16_t> samples(32 * 2, 789);
+  const auto bytes =
+      static_cast<opensl_abi::Uint32>(samples.size() * sizeof(samples.front()));
+  ASSERT_EQ((*queue)->Enqueue(queue, samples.data(), bytes),
+            opensl_abi::kResultSuccess);
+
+  DelayedBufferRelease release(controlled_view);
+  ASSERT_TRUE(release.WaitUntilBlocked());
+  ASSERT_EQ(controlled_view->buffer_count(), 0u);
+  ASSERT_EQ((*queue)->Clear(queue), opensl_abi::kResultSuccess);
+  ASSERT_EQ((*queue)->Enqueue(queue, samples.data(), bytes),
+            opensl_abi::kResultSuccess);
+  release.AllowRelease();
+  release.Join();
+
+  opensl_abi::AndroidSimpleBufferQueueState queue_state{};
+  ASSERT_EQ((*queue)->GetState(queue, &queue_state),
+            opensl_abi::kResultSuccess);
+  EXPECT_EQ(queue_state.count, 1u);
+  EXPECT_EQ(queue_state.index, 0u);
+  const auto before_release = adapter->GetStats();
+  EXPECT_EQ(before_release.submitted_buffers, 2u);
+  EXPECT_EQ(before_release.pending_buffers, 1u);
+  EXPECT_EQ(before_release.consumed_buffers, 0u);
+  EXPECT_EQ(before_release.discarded_buffers, 1u);
+  EXPECT_EQ(before_release.consumed_bytes, 0u);
+
+  controlled_view->ReleaseOne();
+  {
+    std::unique_lock<std::mutex> lock(probe.mutex);
+    EXPECT_TRUE(probe.cv.wait_for(lock, 2s, [&probe] { return probe.calls >= 1; }));
+  }
+  adapter->Shutdown();
+  {
+    std::lock_guard<std::mutex> lock(probe.mutex);
+    EXPECT_EQ(probe.calls, 1);
+  }
+  ASSERT_EQ((*queue)->GetState(queue, &queue_state),
+            opensl_abi::kResultSuccess);
+  EXPECT_EQ(queue_state.count, 0u);
+  EXPECT_EQ(queue_state.index, 1u);
+  const auto after_release = adapter->GetStats();
+  EXPECT_EQ(after_release.pending_buffers, 0u);
+  EXPECT_EQ(after_release.consumed_buffers, 1u);
+  EXPECT_EQ(after_release.discarded_buffers, 1u);
+  EXPECT_EQ(after_release.consumed_bytes, bytes);
+  EXPECT_TRUE(adapter->last_error().ok());
+}
+
+TEST(OpenSlQueueTest, ClearCanRaceBorrowedBufferRelease) {
+  auto controlled = std::make_unique<ControlledAudioSink>();
+  ControlledAudioSink* controlled_view = controlled.get();
+  OpenSlSimpleBufferQueueOptions options;
+  options.max_buffers = 1;
+  std::unique_ptr<OpenSlSimpleBufferQueueAdapter> adapter;
+  ASSERT_TRUE(OpenSlSimpleBufferQueueAdapter::Create(std::move(controlled),
+                                                     options, &adapter)
+                  .ok());
+  auto queue = adapter->interface();
+  const std::vector<std::int16_t> samples(32 * 2, 789);
+  const auto bytes =
+      static_cast<opensl_abi::Uint32>(samples.size() * sizeof(samples.front()));
+  constexpr std::uint64_t kIterations = 64;
+  for (std::uint64_t iteration = 0; iteration < kIterations; ++iteration) {
+    ASSERT_EQ((*queue)->Enqueue(queue, samples.data(), bytes),
+              opensl_abi::kResultSuccess);
+    DelayedBufferRelease release(controlled_view);
+    ASSERT_TRUE(release.WaitUntilBlocked());
+    // Clear and ReleaseBuffer both run after this gate. Joining only after
+    // Clear avoids ordering the ticket reset before the callback's first read.
+    release.AllowRelease();
+    ASSERT_EQ((*queue)->Clear(queue), opensl_abi::kResultSuccess);
+    release.Join();
+    EXPECT_EQ(adapter->GetStats().pending_buffers, 0u);
+  }
+  adapter->Shutdown();
+  const auto stats = adapter->GetStats();
+  EXPECT_EQ(stats.submitted_buffers, kIterations);
+  EXPECT_EQ(stats.consumed_buffers + stats.discarded_buffers, kIterations);
+  EXPECT_EQ(stats.pending_buffers, 0u);
+  EXPECT_TRUE(adapter->last_error().ok());
+}
+
+TEST(OpenSlQueueTest, ShutdownWaitsForDelayedReleaseAfterClear) {
+  auto controlled = std::make_unique<ControlledAudioSink>();
+  ControlledAudioSink* controlled_view = controlled.get();
+  OpenSlSimpleBufferQueueOptions options;
+  options.max_buffers = 1;
+  std::unique_ptr<OpenSlSimpleBufferQueueAdapter> adapter;
+  ASSERT_TRUE(OpenSlSimpleBufferQueueAdapter::Create(std::move(controlled),
+                                                     options, &adapter)
+                  .ok());
+  auto queue = adapter->interface();
+  const std::vector<std::int16_t> samples(32 * 2, 789);
+  const auto bytes =
+      static_cast<opensl_abi::Uint32>(samples.size() * sizeof(samples.front()));
+  ASSERT_EQ((*queue)->Enqueue(queue, samples.data(), bytes),
+            opensl_abi::kResultSuccess);
+  DelayedBufferRelease release(controlled_view);
+  ASSERT_TRUE(release.WaitUntilBlocked());
+  ASSERT_EQ((*queue)->Clear(queue), opensl_abi::kResultSuccess);
+  ASSERT_EQ((*queue)->Enqueue(queue, samples.data(), bytes),
+            opensl_abi::kResultSuccess);
+
+  std::atomic<bool> shutdown_done{false};
+  std::thread shutdown_thread([&] {
+    adapter->Shutdown();
+    shutdown_done.store(true);
+  });
+  EXPECT_TRUE(controlled_view->WaitUntilShutdownStarts());
+  EXPECT_FALSE(shutdown_done.load());
+  release.AllowRelease();
+  release.Join();
+  shutdown_thread.join();
+  EXPECT_TRUE(shutdown_done.load());
+  EXPECT_EQ((*queue)->Enqueue(queue, samples.data(), bytes),
+            opensl_abi::kResultPreconditionsViolated);
+  const auto stats = adapter->GetStats();
+  EXPECT_EQ(stats.pending_buffers, 0u);
+  EXPECT_EQ(stats.consumed_buffers, 0u);
+  EXPECT_EQ(stats.discarded_buffers, 2u);
+  EXPECT_TRUE(adapter->last_error().ok());
+}
+
+TEST(OpenSlQueueTest, FailedEnqueuePreservesCapacityAndCounters) {
+  auto controlled = std::make_unique<ControlledAudioSink>();
+  ControlledAudioSink* controlled_view = controlled.get();
+  OpenSlSimpleBufferQueueOptions options;
+  options.max_buffers = 1;
+  std::unique_ptr<OpenSlSimpleBufferQueueAdapter> adapter;
+  ASSERT_TRUE(OpenSlSimpleBufferQueueAdapter::Create(std::move(controlled),
+                                                     options, &adapter)
+                  .ok());
+  auto queue = adapter->interface();
+  const std::vector<std::int16_t> samples(32 * 2, 789);
+  const auto bytes =
+      static_cast<opensl_abi::Uint32>(samples.size() * sizeof(samples.front()));
+  controlled_view->RejectNextEnqueue();
+  EXPECT_EQ((*queue)->Enqueue(queue, samples.data(), bytes),
+            opensl_abi::kResultResourceError);
+  EXPECT_EQ(adapter->GetStats().submitted_buffers, 0u);
+  EXPECT_EQ(adapter->GetStats().pending_buffers, 0u);
+  ASSERT_EQ((*queue)->Enqueue(queue, samples.data(), bytes),
+            opensl_abi::kResultSuccess);
+  controlled_view->ReleaseOne();
+  const auto stats = adapter->GetStats();
+  EXPECT_EQ(stats.submitted_buffers, 1u);
+  EXPECT_EQ(stats.consumed_buffers, 1u);
+  EXPECT_EQ(stats.discarded_buffers, 0u);
+  EXPECT_EQ(stats.pending_buffers, 0u);
+}
+
+TEST(OpenSlQueueTest, EnqueueCanReleaseBorrowedBufferSynchronously) {
+  auto controlled = std::make_unique<ControlledAudioSink>(true);
+  OpenSlSimpleBufferQueueOptions options;
+  options.max_buffers = 1;
+  CallbackProbe probe;
+  std::unique_ptr<OpenSlSimpleBufferQueueAdapter> adapter;
+  ASSERT_TRUE(OpenSlSimpleBufferQueueAdapter::Create(std::move(controlled),
+                                                     options, &adapter)
+                  .ok());
+  auto queue = adapter->interface();
+  ASSERT_EQ((*queue)->RegisterCallback(queue, OpenSlQueueCallback, &probe),
+            opensl_abi::kResultSuccess);
+  const std::vector<std::int16_t> samples(32 * 2, 789);
+  const auto bytes =
+      static_cast<opensl_abi::Uint32>(samples.size() * sizeof(samples.front()));
+  for (int enqueue = 0; enqueue < 2; ++enqueue) {
+    ASSERT_EQ((*queue)->Enqueue(queue, samples.data(), bytes),
+              opensl_abi::kResultSuccess);
+    EXPECT_EQ(adapter->GetStats().pending_buffers, 0u);
+  }
+  {
+    std::unique_lock<std::mutex> lock(probe.mutex);
+    EXPECT_TRUE(probe.cv.wait_for(lock, 2s, [&probe] { return probe.calls == 2; }));
+  }
+  adapter->Shutdown();
+  const auto stats = adapter->GetStats();
+  EXPECT_EQ(stats.submitted_buffers, 2u);
+  EXPECT_EQ(stats.consumed_buffers, 2u);
+  EXPECT_EQ(stats.discarded_buffers, 0u);
+  EXPECT_EQ(stats.consumed_bytes, 2u * bytes);
+  EXPECT_TRUE(adapter->last_error().ok());
+}
+
 TEST(OpenSlQueueTest, RejectsOverflowAndShutsDownFailClosed) {
   auto controlled = std::make_unique<ControlledAudioSink>();
   OpenSlSimpleBufferQueueOptions options;
@@ -590,12 +876,12 @@ TEST(OpenSlQueueTest, ShutdownWaitsForInFlightCallback) {
 TEST(OpenSlQueueTest, GetStatePollsWhilePlaybackReleasesBuffers) {
   auto controlled = std::make_unique<ControlledAudioSink>();
   ControlledAudioSink* controlled_view = controlled.get();
+  CallbackProbe probe;
   std::unique_ptr<OpenSlSimpleBufferQueueAdapter> adapter;
   ASSERT_TRUE(OpenSlSimpleBufferQueueAdapter::Create(std::move(controlled), {},
                                                      &adapter)
                   .ok());
   auto queue = adapter->interface();
-  CallbackProbe probe;
   ASSERT_EQ((*queue)->RegisterCallback(queue, OpenSlQueueCallback, &probe),
             opensl_abi::kResultSuccess);
 

tests/jnivm_test.cc

@@ -1088,6 +1088,279 @@ TEST_F(JniVmTest, FloatArrayElementsAreMutable) {
   env->ReleaseFloatArrayElements(array, elements, 0);
 }
 
+TEST_F(JniVmTest, ObjectArrayInitializerSurvivesLocalDeletionAndSlotReuse) {
+  JNIEnv *env = vm_->GetJNIEnv();
+  jclass string_class = env->FindClass("java/lang/String");
+  jstring initial = env->NewStringUTF("original initializer");
+  ASSERT_NE(initial, nullptr);
+  jobjectArray array = env->NewObjectArray(1, string_class, initial);
+  ASSERT_NE(array, nullptr);
+  env->DeleteLocalRef(initial);
+
+  jstring replacement = env->NewStringUTF("new allocation");
+  ASSERT_NE(replacement, nullptr);
+  auto element = static_cast<jstring>(env->GetObjectArrayElement(array, 0));
+  ASSERT_NE(element, nullptr);
+  EXPECT_NE(element, replacement);
+  EXPECT_EQ(ReadJavaString(env, element), "original initializer");
+  env->DeleteLocalRef(element);
+  env->DeleteLocalRef(array);
+  env->DeleteLocalRef(replacement);
+  env->DeleteLocalRef(string_class);
+}
+
+TEST_F(JniVmTest, ObjectArrayDeletionReleasesSetElementSlot) {
+  JNIEnv *env = vm_->GetJNIEnv();
+  jclass string_class = env->FindClass("java/lang/String");
+  jobjectArray array = env->NewObjectArray(1, string_class, nullptr);
+  ASSERT_NE(array, nullptr);
+  jstring element = env->NewStringUTF("held by array");
+  ASSERT_NE(element, nullptr);
+  env->SetObjectArrayElement(array, 0, element);
+  env->DeleteLocalRef(element);
+  env->DeleteLocalRef(array);
+
+  // Only compare the saved handle value; never call JNI with a deleted ref.
+  jstring replacement = env->NewStringUTF("reuses released slot");
+  ASSERT_NE(replacement, nullptr);
+  EXPECT_EQ(replacement, element);
+  EXPECT_EQ(ReadJavaString(env, replacement), "reuses released slot");
+  env->DeleteLocalRef(replacement);
+  env->DeleteLocalRef(string_class);
+}
+
+TEST_F(JniVmTest, ObjectArrayInitializerSurvivesLocalFramePop) {
+  JNIEnv *env = vm_->GetJNIEnv();
+  jclass string_class = env->FindClass("java/lang/String");
+  ASSERT_EQ(env->PushLocalFrame(4), JNI_OK);
+  jstring initial = env->NewStringUTF("frame initializer");
+  ASSERT_NE(initial, nullptr);
+  jobjectArray array = env->NewObjectArray(1, string_class, initial);
+  ASSERT_NE(array, nullptr);
+  array = static_cast<jobjectArray>(env->PopLocalFrame(array));
+  ASSERT_NE(array, nullptr);
+
+  jstring replacement = env->NewStringUTF("after frame pop");
+  ASSERT_NE(replacement, nullptr);
+  auto element = static_cast<jstring>(env->GetObjectArrayElement(array, 0));
+  ASSERT_NE(element, nullptr);
+  EXPECT_NE(element, replacement);
+  EXPECT_EQ(ReadJavaString(env, element), "frame initializer");
+  env->DeleteLocalRef(element);
+  env->DeleteLocalRef(array);
+  jstring recycled = env->NewStringUTF("released frame initializer");
+  ASSERT_NE(recycled, nullptr);
+  EXPECT_EQ(recycled, initial);
+  env->DeleteLocalRef(recycled);
+  env->DeleteLocalRef(replacement);
+  env->DeleteLocalRef(string_class);
+}
+
+TEST_F(JniVmTest, ObjectArrayRepeatedInitializerOwnsEveryPosition) {
+  JNIEnv *env = vm_->GetJNIEnv();
+  jclass string_class = env->FindClass("java/lang/String");
+  jstring initial = env->NewStringUTF("three retained positions");
+  ASSERT_NE(initial, nullptr);
+  jobjectArray array = env->NewObjectArray(3, string_class, initial);
+  ASSERT_NE(array, nullptr);
+  env->DeleteLocalRef(initial);
+  env->SetObjectArrayElement(array, 0, nullptr);
+  env->SetObjectArrayElement(array, 1, nullptr);
+
+  jstring replacement = env->NewStringUTF("last position still owns original");
+  ASSERT_NE(replacement, nullptr);
+  auto remaining = static_cast<jstring>(env->GetObjectArrayElement(array, 2));
+  ASSERT_NE(remaining, nullptr);
+  EXPECT_NE(remaining, replacement);
+  EXPECT_EQ(ReadJavaString(env, remaining), "three retained positions");
+  env->DeleteLocalRef(remaining);
+  env->SetObjectArrayElement(array, 2, nullptr);
+  jstring recycled = env->NewStringUTF("last position released");
+  ASSERT_NE(recycled, nullptr);
+  EXPECT_EQ(recycled, initial);
+  env->DeleteLocalRef(array);
+  EXPECT_EQ(ReadJavaString(env, recycled), "last position released");
+  env->DeleteLocalRef(recycled);
+  env->DeleteLocalRef(replacement);
+  env->DeleteLocalRef(string_class);
+}
+
+TEST_F(JniVmTest, ObjectArrayReplacementReleasesSameDifferentAndNullElements) {
+  JNIEnv *env = vm_->GetJNIEnv();
+  jclass string_class = env->FindClass("java/lang/String");
+  jobjectArray array = env->NewObjectArray(1, string_class, nullptr);
+  ASSERT_NE(array, nullptr);
+  jstring first = env->NewStringUTF("first");
+  jstring second = env->NewStringUTF("second");
+  ASSERT_NE(first, nullptr);
+  ASSERT_NE(second, nullptr);
+  env->SetObjectArrayElement(array, 0, first);
+  env->SetObjectArrayElement(array, 0, first);
+  env->DeleteLocalRef(first);
+  auto element = static_cast<jstring>(env->GetObjectArrayElement(array, 0));
+  ASSERT_NE(element, nullptr);
+  EXPECT_EQ(ReadJavaString(env, element), "first");
+  env->DeleteLocalRef(element);
+
+  env->SetObjectArrayElement(array, 0, second);
+  jstring first_reuse = env->NewStringUTF("first slot reused");
+  ASSERT_NE(first_reuse, nullptr);
+  EXPECT_EQ(first_reuse, first);
+  env->DeleteLocalRef(second);
+  element = static_cast<jstring>(env->GetObjectArrayElement(array, 0));
+  ASSERT_NE(element, nullptr);
+  EXPECT_EQ(ReadJavaString(env, element), "second");
+  env->DeleteLocalRef(element);
+  env->SetObjectArrayElement(array, 0, nullptr);
+  EXPECT_EQ(env->GetObjectArrayElement(array, 0), nullptr);
+  jstring second_reuse = env->NewStringUTF("second slot reused");
+  ASSERT_NE(second_reuse, nullptr);
+  EXPECT_EQ(second_reuse, second);
+  env->DeleteLocalRef(array);
+  env->DeleteLocalRef(first_reuse);
+  env->DeleteLocalRef(second_reuse);
+  env->DeleteLocalRef(string_class);
+}
+
+TEST_F(JniVmTest, ObjectArrayDeletionRecursivelyReleasesNestedElements) {
+  JNIEnv *env = vm_->GetJNIEnv();
+  jclass string_class = env->FindClass("java/lang/String");
+  jclass object_class = env->FindClass("java/lang/Object");
+  jstring initial = env->NewStringUTF("nested string");
+  ASSERT_NE(initial, nullptr);
+  jobjectArray inner = env->NewObjectArray(1, string_class, initial);
+  ASSERT_NE(inner, nullptr);
+  jobjectArray outer = env->NewObjectArray(1, object_class, inner);
+  ASSERT_NE(outer, nullptr);
+  env->DeleteLocalRef(initial);
+  env->DeleteLocalRef(inner);
+
+  auto retained_inner =
+      static_cast<jobjectArray>(env->GetObjectArrayElement(outer, 0));
+  ASSERT_NE(retained_inner, nullptr);
+  auto element =
+      static_cast<jstring>(env->GetObjectArrayElement(retained_inner, 0));
+  ASSERT_NE(element, nullptr);
+  EXPECT_EQ(ReadJavaString(env, element), "nested string");
+  env->DeleteLocalRef(element);
+  env->DeleteLocalRef(retained_inner);
+  env->DeleteLocalRef(outer);
+  jstring recycled = env->NewStringUTF("recursive release completed");
+  ASSERT_NE(recycled, nullptr);
+  EXPECT_EQ(recycled, initial);
+  env->DeleteLocalRef(recycled);
+  env->DeleteLocalRef(string_class);
+  env->DeleteLocalRef(object_class);
+}
+
+TEST_F(JniVmTest, ObjectArrayNullAndEmptyArraysDoNotRetainInitializer) {
+  JNIEnv *env = vm_->GetJNIEnv();
+  jclass string_class = env->FindClass("java/lang/String");
+  jstring initial = env->NewStringUTF("not stored in an empty array");
+  ASSERT_NE(initial, nullptr);
+  jobjectArray empty = env->NewObjectArray(0, string_class, initial);
+  jobjectArray nulls = env->NewObjectArray(3, string_class, nullptr);
+  ASSERT_NE(empty, nullptr);
+  ASSERT_NE(nulls, nullptr);
+  EXPECT_EQ(env->GetArrayLength(empty), 0);
+  EXPECT_EQ(env->GetArrayLength(nulls), 3);
+  for (jsize index = 0; index < 3; ++index) {
+    EXPECT_EQ(env->GetObjectArrayElement(nulls, index), nullptr);
+    env->SetObjectArrayElement(nulls, index, nullptr);
+  }
+  env->DeleteLocalRef(initial);
+  jstring recycled = env->NewStringUTF("empty initializer was released");
+  ASSERT_NE(recycled, nullptr);
+  EXPECT_EQ(recycled, initial);
+  env->DeleteLocalRef(empty);
+  env->DeleteLocalRef(nulls);
+  EXPECT_EQ(ReadJavaString(env, recycled), "empty initializer was released");
+  env->DeleteLocalRef(recycled);
+  env->DeleteLocalRef(string_class);
+}
+
+TEST_F(JniVmTest, ObjectArraySelfReferenceCanBeBrokenWithValidExternalReference) {
+  JNIEnv *env = vm_->GetJNIEnv();
+  jclass object_class = env->FindClass("java/lang/Object");
+  jobjectArray array = env->NewObjectArray(2, object_class, nullptr);
+  ASSERT_NE(array, nullptr);
+  jstring initial = env->NewStringUTF("owned beside self reference");
+  ASSERT_NE(initial, nullptr);
+  env->SetObjectArrayElement(array, 0, array);
+  env->SetObjectArrayElement(array, 1, initial);
+  auto external = static_cast<jobjectArray>(env->NewGlobalRef(array));
+  ASSERT_NE(external, nullptr);
+  env->DeleteLocalRef(array);
+  env->DeleteLocalRef(initial);
+
+  jobject self = env->GetObjectArrayElement(external, 0);
+  ASSERT_NE(self, nullptr);
+  EXPECT_EQ(env->IsSameObject(self, external), JNI_TRUE);
+  env->DeleteLocalRef(self);
+  env->SetObjectArrayElement(external, 0, nullptr);
+  auto element = static_cast<jstring>(env->GetObjectArrayElement(external, 1));
+  ASSERT_NE(element, nullptr);
+  EXPECT_EQ(ReadJavaString(env, element), "owned beside self reference");
+  env->DeleteLocalRef(element);
+  env->DeleteGlobalRef(external);
+  jstring recycled = env->NewStringUTF("broken cycle released its element");
+  ASSERT_NE(recycled, nullptr);
+  EXPECT_EQ(recycled, initial);
+  env->DeleteLocalRef(recycled);
+  env->DeleteLocalRef(object_class);
+}
+
+TEST_F(JniVmTest, ObjectArraySurvivesTemporaryVmTeardownAndEnvironmentReattachment) {
+  JNIEnv *env = vm_->GetJNIEnv();
+  jclass string_class = env->FindClass("java/lang/String");
+  jstring initial = env->NewStringUTF("surviving VM array");
+  ASSERT_NE(initial, nullptr);
+  jobjectArray array = env->NewObjectArray(1, string_class, initial);
+  ASSERT_NE(array, nullptr);
+  env->DeleteLocalRef(initial);
+  {
+    VM temporary;
+    ASSERT_NE(temporary.GetJNIEnv(), nullptr);
+  }
+  // Temporary VM teardown invalidates the thread's environment; reacquire it
+  // before invoking JNI on references owned by the surviving VM.
+  env = vm_->GetJNIEnv();
+  ASSERT_NE(env, nullptr);
+  auto element = static_cast<jstring>(env->GetObjectArrayElement(array, 0));
+  ASSERT_NE(element, nullptr);
+  EXPECT_EQ(ReadJavaString(env, element), "surviving VM array");
+  env->DeleteLocalRef(element);
+  env->DeleteLocalRef(array);
+  jstring recycled = env->NewStringUTF("surviving array released");
+  ASSERT_NE(recycled, nullptr);
+  EXPECT_EQ(recycled, initial);
+  env->DeleteLocalRef(recycled);
+  env->DeleteLocalRef(string_class);
+}
+
+TEST_F(JniVmTest, ObjectArrayLiveAtProcessExitDoesNotReenterDestroyedTables) {
+  ASSERT_EXIT(
+      {
+        VM temporary;
+        JNIEnv *env = temporary.GetJNIEnv();
+        if (env == nullptr) std::_Exit(10);
+        jclass string_class = env->FindClass("java/lang/String");
+        jclass object_class = env->FindClass("java/lang/Object");
+        jstring initial = env->NewStringUTF("live at shutdown");
+        if (initial == nullptr) std::_Exit(11);
+        jobjectArray inner = env->NewObjectArray(3, string_class, initial);
+        if (inner == nullptr) std::_Exit(12);
+        jobjectArray outer = env->NewObjectArray(1, object_class, inner);
+        if (outer == nullptr) std::_Exit(13);
+        env->DeleteLocalRef(initial);
+        env->DeleteLocalRef(inner);
+        // Run global destructors with outer/inner/string still alive. Release
+        // of elements belongs to explicit reference release, not ~PseudoArray.
+        std::exit(0);
+      },
+      ::testing::ExitedWithCode(0), "");
+}
+
 constexpr int kMoreThanHandleCapacity = 150000;
 
 TEST_F(JniVmTest, RepeatedFindClassDoesNotExhaustHandles) {