From 4839a032045c52c8fcbc6770a1b99d51bfc87e0e Mon Sep 17 00:00:00 2001
From: Maciej Mensfeld <maciej@mensfeld.pl>
Date: Mon, 21 Nov 2022 12:01:14 +0100
Subject: [PATCH] Kind fetcher locks are not fully thread-safe

In case of extensive concurrent usage, the mutex handed over to two threads under same key may differ as illustrated below:

```ruby
require 'concurrent'

10000.times do
  kind_fetcher_locks = Concurrent::Hash.new { |hash, key| hash[key] = Mutex.new }

  refs = Set.new
  100.times.map do |i|
    Thread.new { refs << kind_fetcher_locks[i % 50].object_id }
  end.each(&:join)

  raise "Not 50 but #{refs.count}" unless refs.size == 50
end
```

this can lead to really weird issues.

Works when fixed as above:

```ruby
require 'concurrent'

10000.times do
  mutex = Mutex.new
  # kind_fetcher_locks = Concurrent::Hash.new { |hash, key| hash[key] = Mutex.new }

  kind_fetcher_locks = Concurrent::Hash.new do |hash, key|
    mutex.synchronize do
      break hash[key] if hash.key?(key)
      hash[key] = Mutex.new
    end
  end

  refs = Set.new
  100.times.map do |i|
    Thread.new { refs << kind_fetcher_locks[i % 50].object_id }
  end.each(&:join)

  raise "Not 50 but #{refs.count}" unless refs.size == 50
end
```

ref https://github.com/ruby-concurrency/concurrent-ruby/issues/970
---
 lib/krane/resource_cache.rb | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/lib/krane/resource_cache.rb b/lib/krane/resource_cache.rb
index baba9f8a8..87d0a87a4 100644
--- a/lib/krane/resource_cache.rb
+++ b/lib/krane/resource_cache.rb
@@ -8,8 +8,13 @@ class ResourceCache
 
     def initialize(task_config)
       @task_config = task_config
-
-      @kind_fetcher_locks = Concurrent::Hash.new { |hash, key| hash[key] = Mutex.new }
+      @mutex = Mutex.new
+      @kind_fetcher_locks = Concurrent::Hash.new do |hash, key|
+        @mutex.synchronize do
+          break hash[key] if hash.key?(key)
+          hash[key] = Mutex.new
+        end
+      end
       @data = Concurrent::Hash.new
       @kubectl = Kubectl.new(task_config: @task_config, log_failure_by_default: false)
     end