this repo has no description
Surface error when all storage pids are unknown for a shard
+14
-7
+14
-7
lib/transaction.ex
+14
-7
lib/transaction.ex
···
116
116
# Successful read, return results immediately
117
117
{:ok, _map} = read_result -> {:halt, read_result}
118
118
# Shard information is outdated, retry get_shards
119
+
{:error, :storage_pids_unknown} = error -> {:cont, nil, error}
119
120
{:error, :wrong_server} = error -> {:cont, get_shards.(), error}
120
121
# These are retryable, retry with the same shards
121
122
{:error, :timeout} = error -> {:cont, shard_result, error}
···
149
150
|> Enum.sort_by(fn {{ids, _pids}, _keys} -> ids end)
150
151
# Send read requests in parallel
151
152
|> Enum.map(fn {{_ids, pids}, keys} ->
152
-
# TODO: handle all pids being unknown (nil)
153
-
{:ok, pid} = random_not_nil(pids)
154
-
req_id = Storage.read_send(pid, txn.generation, txn.read_version, keys)
155
-
{keys, req_id}
153
+
case random_not_nil(pids) do
154
+
{:ok, pid} ->
155
+
req_id = Storage.read_send(pid, txn.generation, txn.read_version, keys)
156
+
{keys, req_id}
157
+
:error ->
158
+
{keys, {:error, :storage_pids_unknown}}
159
+
end
156
160
end)
157
-
|> Enum.map(fn {keys, req_id} ->
158
-
values = Storage.read_receive(req_id)
159
-
{keys, values}
161
+
|> Enum.map(fn
162
+
{_keys, {:error, _err}} = tuple ->
163
+
tuple
164
+
{keys, req_id} ->
165
+
values = Storage.read_receive(req_id)
166
+
{keys, values}
160
167
end)
161
168
# Coalesce any errors
162
169
|> Enum.reduce_while([], fn {keys, values}, acc ->