Update to use LibAwsHTTP

JuliaWeb · Jun 13, 2024 · 87dd0dc · 87dd0dc
1 parent a94e9d8
commit 87dd0dc
Show file tree

Hide file tree

Showing 8 changed files with 563 additions and 296 deletions.
diff --git a/Project.toml b/Project.toml
@@ -4,11 +4,11 @@ version = "1.0.0"
 
 [deps]
 CodecZlib = "944b1d66-785c-5afd-91f1-9de20f533193"
+LibAwsCommon = "c6e421ba-b5f8-4792-a1c4-42948de3ed9d"
+LibAwsHTTP = "ce851869-0d7a-41e7-95ef-2d4cb63876dd"
+LibAwsIO = "a5388770-19df-4151-b103-3d71de896ddf"
 Mmap = "a63ad114-7e13-5084-954f-fe012c677804"
 URIs = "5c2747f8-b7ea-4ff2-ba2e-563bfd36b1d4"
-aws_c_common_jll = "73048d1d-b8c4-5092-a58d-866c5e8d1e50"
-aws_c_http_jll = "3254fc65-9028-534d-aa9d-d76d128babc6"
-aws_c_io_jll = "13c41daa-f319-5298-b5eb-5754e0170d52"
 
 [compat]
 julia = "1.9"

diff --git a/src/HTTP2.jl b/src/HTTP2.jl
diff --git a/src/c.jl b/src/c.jl
@@ -1,155 +1,3 @@
-@enum aws_io_errors::UInt32 begin
-    AWS_IO_CHANNEL_ERROR_ERROR_CANT_ACCEPT_INPUT = 1024
-    AWS_IO_CHANNEL_UNKNOWN_MESSAGE_TYPE = 1025
-    AWS_IO_CHANNEL_READ_WOULD_EXCEED_WINDOW = 1026
-    AWS_IO_EVENT_LOOP_ALREADY_ASSIGNED = 1027
-    AWS_IO_EVENT_LOOP_SHUTDOWN = 1028
-    AWS_IO_TLS_ERROR_NEGOTIATION_FAILURE = 1029
-    AWS_IO_TLS_ERROR_NOT_NEGOTIATED = 1030
-    AWS_IO_TLS_ERROR_WRITE_FAILURE = 1031
-    AWS_IO_TLS_ERROR_ALERT_RECEIVED = 1032
-    AWS_IO_TLS_CTX_ERROR = 1033
-    AWS_IO_TLS_VERSION_UNSUPPORTED = 1034
-    AWS_IO_TLS_CIPHER_PREF_UNSUPPORTED = 1035
-    AWS_IO_MISSING_ALPN_MESSAGE = 1036
-    AWS_IO_UNHANDLED_ALPN_PROTOCOL_MESSAGE = 1037
-    AWS_IO_FILE_VALIDATION_FAILURE = 1038
-    AWS_ERROR_IO_EVENT_LOOP_THREAD_ONLY = 1039
-    AWS_ERROR_IO_ALREADY_SUBSCRIBED = 1040
-    AWS_ERROR_IO_NOT_SUBSCRIBED = 1041
-    AWS_ERROR_IO_OPERATION_CANCELLED = 1042
-    AWS_IO_READ_WOULD_BLOCK = 1043
-    AWS_IO_BROKEN_PIPE = 1044
-    AWS_IO_SOCKET_UNSUPPORTED_ADDRESS_FAMILY = 1045
-    AWS_IO_SOCKET_INVALID_OPERATION_FOR_TYPE = 1046
-    AWS_IO_SOCKET_CONNECTION_REFUSED = 1047
-    AWS_IO_SOCKET_TIMEOUT = 1048
-    AWS_IO_SOCKET_NO_ROUTE_TO_HOST = 1049
-    AWS_IO_SOCKET_NETWORK_DOWN = 1050
-    AWS_IO_SOCKET_CLOSED = 1051
-    AWS_IO_SOCKET_NOT_CONNECTED = 1052
-    AWS_IO_SOCKET_INVALID_OPTIONS = 1053
-    AWS_IO_SOCKET_ADDRESS_IN_USE = 1054
-    AWS_IO_SOCKET_INVALID_ADDRESS = 1055
-    AWS_IO_SOCKET_ILLEGAL_OPERATION_FOR_STATE = 1056
-    AWS_IO_SOCKET_CONNECT_ABORTED = 1057
-    AWS_IO_DNS_QUERY_FAILED = 1058
-    AWS_IO_DNS_INVALID_NAME = 1059
-    AWS_IO_DNS_NO_ADDRESS_FOR_HOST = 1060
-    AWS_IO_DNS_HOST_REMOVED_FROM_CACHE = 1061
-    AWS_IO_STREAM_INVALID_SEEK_POSITION = 1062
-    AWS_IO_STREAM_READ_FAILED = 1063
-    DEPRECATED_AWS_IO_INVALID_FILE_HANDLE = 1064
-    AWS_IO_SHARED_LIBRARY_LOAD_FAILURE = 1065
-    AWS_IO_SHARED_LIBRARY_FIND_SYMBOL_FAILURE = 1066
-    AWS_IO_TLS_NEGOTIATION_TIMEOUT = 1067
-    AWS_IO_TLS_ALERT_NOT_GRACEFUL = 1068
-    AWS_IO_MAX_RETRIES_EXCEEDED = 1069
-    AWS_IO_RETRY_PERMISSION_DENIED = 1070
-    AWS_IO_TLS_DIGEST_ALGORITHM_UNSUPPORTED = 1071
-    AWS_IO_TLS_SIGNATURE_ALGORITHM_UNSUPPORTED = 1072
-    AWS_ERROR_PKCS11_VERSION_UNSUPPORTED = 1073
-    AWS_ERROR_PKCS11_TOKEN_NOT_FOUND = 1074
-    AWS_ERROR_PKCS11_KEY_NOT_FOUND = 1075
-    AWS_ERROR_PKCS11_KEY_TYPE_UNSUPPORTED = 1076
-    AWS_ERROR_PKCS11_UNKNOWN_CRYPTOKI_RETURN_VALUE = 1077
-    AWS_ERROR_PKCS11_CKR_CANCEL = 1078
-    AWS_ERROR_PKCS11_CKR_HOST_MEMORY = 1079
-    AWS_ERROR_PKCS11_CKR_SLOT_ID_INVALID = 1080
-    AWS_ERROR_PKCS11_CKR_GENERAL_ERROR = 1081
-    AWS_ERROR_PKCS11_CKR_FUNCTION_FAILED = 1082
-    AWS_ERROR_PKCS11_CKR_ARGUMENTS_BAD = 1083
-    AWS_ERROR_PKCS11_CKR_NO_EVENT = 1084
-    AWS_ERROR_PKCS11_CKR_NEED_TO_CREATE_THREADS = 1085
-    AWS_ERROR_PKCS11_CKR_CANT_LOCK = 1086
-    AWS_ERROR_PKCS11_CKR_ATTRIBUTE_READ_ONLY = 1087
-    AWS_ERROR_PKCS11_CKR_ATTRIBUTE_SENSITIVE = 1088
-    AWS_ERROR_PKCS11_CKR_ATTRIBUTE_TYPE_INVALID = 1089
-    AWS_ERROR_PKCS11_CKR_ATTRIBUTE_VALUE_INVALID = 1090
-    AWS_ERROR_PKCS11_CKR_ACTION_PROHIBITED = 1091
-    AWS_ERROR_PKCS11_CKR_DATA_INVALID = 1092
-    AWS_ERROR_PKCS11_CKR_DATA_LEN_RANGE = 1093
-    AWS_ERROR_PKCS11_CKR_DEVICE_ERROR = 1094
-    AWS_ERROR_PKCS11_CKR_DEVICE_MEMORY = 1095
-    AWS_ERROR_PKCS11_CKR_DEVICE_REMOVED = 1096
-    AWS_ERROR_PKCS11_CKR_ENCRYPTED_DATA_INVALID = 1097
-    AWS_ERROR_PKCS11_CKR_ENCRYPTED_DATA_LEN_RANGE = 1098
-    AWS_ERROR_PKCS11_CKR_FUNCTION_CANCELED = 1099
-    AWS_ERROR_PKCS11_CKR_FUNCTION_NOT_PARALLEL = 1100
-    AWS_ERROR_PKCS11_CKR_FUNCTION_NOT_SUPPORTED = 1101
-    AWS_ERROR_PKCS11_CKR_KEY_HANDLE_INVALID = 1102
-    AWS_ERROR_PKCS11_CKR_KEY_SIZE_RANGE = 1103
-    AWS_ERROR_PKCS11_CKR_KEY_TYPE_INCONSISTENT = 1104
-    AWS_ERROR_PKCS11_CKR_KEY_NOT_NEEDED = 1105
-    AWS_ERROR_PKCS11_CKR_KEY_CHANGED = 1106
-    AWS_ERROR_PKCS11_CKR_KEY_NEEDED = 1107
-    AWS_ERROR_PKCS11_CKR_KEY_INDIGESTIBLE = 1108
-    AWS_ERROR_PKCS11_CKR_KEY_FUNCTION_NOT_PERMITTED = 1109
-    AWS_ERROR_PKCS11_CKR_KEY_NOT_WRAPPABLE = 1110
-    AWS_ERROR_PKCS11_CKR_KEY_UNEXTRACTABLE = 1111
-    AWS_ERROR_PKCS11_CKR_MECHANISM_INVALID = 1112
-    AWS_ERROR_PKCS11_CKR_MECHANISM_PARAM_INVALID = 1113
-    AWS_ERROR_PKCS11_CKR_OBJECT_HANDLE_INVALID = 1114
-    AWS_ERROR_PKCS11_CKR_OPERATION_ACTIVE = 1115
-    AWS_ERROR_PKCS11_CKR_OPERATION_NOT_INITIALIZED = 1116
-    AWS_ERROR_PKCS11_CKR_PIN_INCORRECT = 1117
-    AWS_ERROR_PKCS11_CKR_PIN_INVALID = 1118
-    AWS_ERROR_PKCS11_CKR_PIN_LEN_RANGE = 1119
-    AWS_ERROR_PKCS11_CKR_PIN_EXPIRED = 1120
-    AWS_ERROR_PKCS11_CKR_PIN_LOCKED = 1121
-    AWS_ERROR_PKCS11_CKR_SESSION_CLOSED = 1122
-    AWS_ERROR_PKCS11_CKR_SESSION_COUNT = 1123
-    AWS_ERROR_PKCS11_CKR_SESSION_HANDLE_INVALID = 1124
-    AWS_ERROR_PKCS11_CKR_SESSION_PARALLEL_NOT_SUPPORTED = 1125
-    AWS_ERROR_PKCS11_CKR_SESSION_READ_ONLY = 1126
-    AWS_ERROR_PKCS11_CKR_SESSION_EXISTS = 1127
-    AWS_ERROR_PKCS11_CKR_SESSION_READ_ONLY_EXISTS = 1128
-    AWS_ERROR_PKCS11_CKR_SESSION_READ_WRITE_SO_EXISTS = 1129
-    AWS_ERROR_PKCS11_CKR_SIGNATURE_INVALID = 1130
-    AWS_ERROR_PKCS11_CKR_SIGNATURE_LEN_RANGE = 1131
-    AWS_ERROR_PKCS11_CKR_TEMPLATE_INCOMPLETE = 1132
-    AWS_ERROR_PKCS11_CKR_TEMPLATE_INCONSISTENT = 1133
-    AWS_ERROR_PKCS11_CKR_TOKEN_NOT_PRESENT = 1134
-    AWS_ERROR_PKCS11_CKR_TOKEN_NOT_RECOGNIZED = 1135
-    AWS_ERROR_PKCS11_CKR_TOKEN_WRITE_PROTECTED = 1136
-    AWS_ERROR_PKCS11_CKR_UNWRAPPING_KEY_HANDLE_INVALID = 1137
-    AWS_ERROR_PKCS11_CKR_UNWRAPPING_KEY_SIZE_RANGE = 1138
-    AWS_ERROR_PKCS11_CKR_UNWRAPPING_KEY_TYPE_INCONSISTENT = 1139
-    AWS_ERROR_PKCS11_CKR_USER_ALREADY_LOGGED_IN = 1140
-    AWS_ERROR_PKCS11_CKR_USER_NOT_LOGGED_IN = 1141
-    AWS_ERROR_PKCS11_CKR_USER_PIN_NOT_INITIALIZED = 1142
-    AWS_ERROR_PKCS11_CKR_USER_TYPE_INVALID = 1143
-    AWS_ERROR_PKCS11_CKR_USER_ANOTHER_ALREADY_LOGGED_IN = 1144
-    AWS_ERROR_PKCS11_CKR_USER_TOO_MANY_TYPES = 1145
-    AWS_ERROR_PKCS11_CKR_WRAPPED_KEY_INVALID = 1146
-    AWS_ERROR_PKCS11_CKR_WRAPPED_KEY_LEN_RANGE = 1147
-    AWS_ERROR_PKCS11_CKR_WRAPPING_KEY_HANDLE_INVALID = 1148
-    AWS_ERROR_PKCS11_CKR_WRAPPING_KEY_SIZE_RANGE = 1149
-    AWS_ERROR_PKCS11_CKR_WRAPPING_KEY_TYPE_INCONSISTENT = 1150
-    AWS_ERROR_PKCS11_CKR_RANDOM_SEED_NOT_SUPPORTED = 1151
-    AWS_ERROR_PKCS11_CKR_RANDOM_NO_RNG = 1152
-    AWS_ERROR_PKCS11_CKR_DOMAIN_PARAMS_INVALID = 1153
-    AWS_ERROR_PKCS11_CKR_CURVE_NOT_SUPPORTED = 1154
-    AWS_ERROR_PKCS11_CKR_BUFFER_TOO_SMALL = 1155
-    AWS_ERROR_PKCS11_CKR_SAVED_STATE_INVALID = 1156
-    AWS_ERROR_PKCS11_CKR_INFORMATION_SENSITIVE = 1157
-    AWS_ERROR_PKCS11_CKR_STATE_UNSAVEABLE = 1158
-    AWS_ERROR_PKCS11_CKR_CRYPTOKI_NOT_INITIALIZED = 1159
-    AWS_ERROR_PKCS11_CKR_CRYPTOKI_ALREADY_INITIALIZED = 1160
-    AWS_ERROR_PKCS11_CKR_MUTEX_BAD = 1161
-    AWS_ERROR_PKCS11_CKR_MUTEX_NOT_LOCKED = 1162
-    AWS_ERROR_PKCS11_CKR_NEW_PIN_MODE = 1163
-    AWS_ERROR_PKCS11_CKR_NEXT_OTP = 1164
-    AWS_ERROR_PKCS11_CKR_EXCEEDED_MAX_ITERATIONS = 1165
-    AWS_ERROR_PKCS11_CKR_FIPS_SELF_TEST_FAILED = 1166
-    AWS_ERROR_PKCS11_CKR_LIBRARY_LOAD_FAILED = 1167
-    AWS_ERROR_PKCS11_CKR_PIN_TOO_WEAK = 1168
-    AWS_ERROR_PKCS11_CKR_PUBLIC_KEY_INVALID = 1169
-    AWS_ERROR_PKCS11_CKR_FUNCTION_REJECTED = 1170
-    AWS_ERROR_IO_PINNED_EVENT_LOOP_MISMATCH = 1171
-    AWS_IO_ERROR_END_RANGE = 2047
-    AWS_IO_INVALID_FILE_HANDLE = 50
-end
 
 const aws_allocator = Cvoid
 
@@ -177,12 +25,7 @@ function aws_error_debug_str(err)
     ccall((:aws_error_debug_str, libaws_c_common), Ptr{Cchar}, (Cint,), err)
 end
 
-struct AWSError <: Exception
-    msg::String
-end
 
-aws_error() = AWSError(unsafe_string(aws_error_debug_str(aws_last_error())))
-aws_throw_error() = throw(aws_error())
 
 @enum aws_log_level::UInt32 begin
     AWS_LL_NONE = 0

diff --git a/src/client.jl b/src/client.jl
@@ -62,7 +62,8 @@ end
 
 const on_setup = Ref{Ptr{Cvoid}}(C_NULL)
 
-function c_on_setup(conn, error_code, ctx)
+function c_on_setup(conn, error_code, ctx_ptr)
+    ctx = unsafe_pointer_to_objref(ctx_ptr)
     # println("on setup")
     if error_code != 0
         ctx.error = CapturedException(aws_error(), Base.backtrace())
@@ -146,9 +147,20 @@ function c_on_setup(conn, error_code, ctx)
     if input_stream != C_NULL
         aws_http_message_set_body_stream(request, input_stream)
     end
-
-    final_request = aws_http_make_request_options(request, ctx)
-    stream = aws_http_connection_make_request(conn, final_request)
+    ctx.request_options = Ref(aws_http_make_request_options(
+        1,
+        request,
+        pointer_from_objref(ctx),
+        on_response_headers[],
+        on_response_header_block_done[],
+        on_response_body[],
+        on_metrics[],
+        on_complete[],
+        on_destroy[],
+        false,
+        0 # response_first_byte_timeout_ms
+    ))
+    stream = aws_http_connection_make_request(conn, ctx.request_options)
     if stream == C_NULL
         ctx.error = CapturedException(aws_error(), Base.backtrace())
         ctx.should_retry = true
@@ -167,13 +179,14 @@ end
 
 const on_shutdown = Ref{Ptr{Cvoid}}(C_NULL)
 
-function c_on_shutdown(conn, error_code, ctx)
+function c_on_shutdown(conn, error_code, ctx_ptr)
     return
 end
 
 const on_response_headers = Ref{Ptr{Cvoid}}(C_NULL)
 
-function c_on_response_headers(stream, header_block, header_array, num_headers, ctx)
+function c_on_response_headers(stream, header_block, header_array, num_headers, ctx_ptr)
+    ctx = unsafe_pointer_to_objref(ctx_ptr)
     headers = unsafe_wrap(Array, Ptr{aws_http_header}(header_array), num_headers)
     for header in headers
         name = unsafe_string(header.name.ptr, header.name.len)
@@ -187,7 +200,8 @@ writebuf(body, maxsize=length(body) == 0 ? typemax(Int64) : length(body)) = IOBu
 
 const on_response_header_block_done = Ref{Ptr{Cvoid}}(C_NULL)
 
-function c_on_response_header_block_done(stream, header_block, ctx)
+function c_on_response_header_block_done(stream, header_block, ctx_ptr)
+    ctx = unsafe_pointer_to_objref(ctx_ptr)
     ref = Ref{Cint}()
     aws_http_stream_get_incoming_response_status(stream, ref)
     ctx.response.status = ref[]
@@ -215,7 +229,7 @@ function c_on_response_header_block_done(stream, header_block, ctx)
     ctx.temp_response_body = if ctx.decompress === true || (ctx.decompress === nothing && getheader(ctx.response.headers, "content-encoding") == "gzip")
         # we're going to gzip decompress the response body
         ctx.gzip_decompressing = true
-        CodecZlibNG.GzipDecompressorStream(response_body isa AbstractVector{UInt8} ? writebuf(response_body, typemax(Int64)) : response_body)
+        CodecZlib.GzipDecompressorStream(response_body isa AbstractVector{UInt8} ? writebuf(response_body, typemax(Int64)) : response_body)
     elseif response_body isa AbstractVector{UInt8}
         writebuf(response_body)
     else
@@ -239,7 +253,8 @@ function hasroom(buf::Base.GenericIOBuffer, n)
     return (requested_buffer_capacity <= length(buf.data)) || (buf.writable && requested_buffer_capacity <= buf.maxsize)
 end
 
-function c_on_response_body(stream, data::Ptr{aws_byte_cursor}, ctx)
+function c_on_response_body(stream, data::Ptr{aws_byte_cursor}, ctx_ptr)
+    ctx = unsafe_pointer_to_objref(ctx_ptr)
     bc = unsafe_load(data)
     body = ctx.temp_response_body
     ctx.response.metrics.response_body_length += bc.len
@@ -258,7 +273,7 @@ function c_on_response_body(stream, data::Ptr{aws_byte_cursor}, ctx)
             return Cint(0)
         end
         unsafe_write(body, bc.ptr, bc.len)
-    elseif body isa CodecZlibNG.TranscodingStreams.TranscodingStream{GzipDecompressor, IOBuffer}
+    elseif body isa CodecZlib.TranscodingStreams.TranscodingStream{GzipDecompressor, IOBuffer}
         unsafe_write(body, bc.ptr, bc.len)
     else
         unsafe_write(body, bc.ptr, bc.len)
@@ -269,7 +284,8 @@ end
 
 const on_metrics = Ref{Ptr{Cvoid}}(C_NULL)
 
-function c_on_metrics(stream, metrics::Ptr{StreamMetrics}, ctx)
+function c_on_metrics(stream, metrics::Ptr{StreamMetrics}, ctx_ptr)
+    ctx = unsafe_pointer_to_objref(ctx_ptr)
     # println("on metrics")
     m = unsafe_load(metrics)
     if m.send_start_timestamp_ns != -1
@@ -280,7 +296,8 @@ end
 
 const on_complete = Ref{Ptr{Cvoid}}(C_NULL)
 
-function c_on_complete(stream, error_code, ctx)
+function c_on_complete(stream, error_code, ctx_ptr)
+    ctx = unsafe_pointer_to_objref(ctx_ptr)
     if ctx.gzip_decompressing
         close(ctx.temp_response_body)
     end
@@ -298,7 +315,7 @@ function c_on_destroy(ctx)
     return
 end
 
-request(method, url, h=Header[], b::RequestBodyTypes=nothing; allocator=ALLOCATOR[], headers=h, body::RequestBodyTypes=b, query=nothing, kw...) =
+request(method, url, h=Header[], b::RequestBodyTypes=nothing; allocator=default_aws_allocator(), headers=h, body::RequestBodyTypes=b, query=nothing, kw...) =
     request(Request(method, url, headers, body, allocator, query); kw...)
 
 # main entrypoint for making an HTTP request
@@ -316,20 +333,22 @@ function request(req::Request;
     decompress::Union{Nothing, Bool}=nothing,
     status_exception::Bool=true,
     retry_non_idempotent::Bool=false,
+    require_ssl_verification::Bool=true,
     modifier=nothing,
     verbose=0,
     # NOTE: new keywords must also be added to the @client macro definition below
 )
 
-    verbose >= 1 && @info "getting client: $(req.uri.scheme), $(req.uri.host), $(req._uri.port)"
-    client = something(client, getclient((req.uri.scheme, req.uri.host, req._uri.port)))::Client
+    verbose >= 1 && @info "getting client: $(req.uri.scheme), $(req.uri.host), $(getport(req._uri))"
+    client = something(client, getclient((req.uri.scheme, req.uri.host, getport(req._uri), require_ssl_verification)))::Client
     # create a request context for shared state that we pass between all the callbacks
     ctx = RequestContext(client, req, Response(response_body), decompress, status_exception, retry_non_idempotent, modifier, verbose)
 
     # NOTE: this is threadsafe based on our usage of the "standard retry strategy" default aws implementation
 @label acquire_retry_token
     verbose >= 1 && @info "acquiring retry token: $(req.uri.host)"
-    if aws_retry_strategy_acquire_retry_token(client.retry_strategy, req._uri.host_name, on_acquired[], ctx, client.retry_timeout_ms) != 0
+    host_ref = Ref(req._uri.host_name)
+    if aws_retry_strategy_acquire_retry_token(client.retry_strategy, host_ref, on_acquired[], pointer_from_objref(ctx), client.retry_timeout_ms) != 0
         ctx.error = CapturedException(aws_error(), Base.backtrace())
         @goto error_fail
     end
@@ -349,9 +368,9 @@ function request(req::Request;
             ctx.request.uri = resolvereference(ctx.request.uri, getheader(ctx.response.headers, "location"))
             ctx.request._uri = aws_uri(ctx.request.uri, client.allocator)
             ctx.request.method = newmethod(ctx.request.method, ctx.response.status, redirect_method)
-            verbose >= 1 && @info "getting redirect client: $(ctx.request.uri.scheme), $(ctx.request.uri.host), $(ctx.request._uri.port)"
+            verbose >= 1 && @info "getting redirect client: $(ctx.request.uri.scheme), $(ctx.request.uri.host), $(getport(ctx.request._uri))"
             old_client = ctx.client
-            ctx.client = getclient((ctx.request.uri.scheme, ctx.request.uri.host, ctx.request._uri.port))
+            ctx.client = getclient((ctx.request.uri.scheme, ctx.request.uri.host, getport(ctx.request._uri)))
             verbose >= 2 && @info "redirecting to $(string(ctx.request.uri)) with method: $(String(ctx.request.method))"
             if ctx.request.method == "GET"
                 ctx.request.body = UInt8[]
@@ -406,7 +425,7 @@ function request(req::Request;
             ctx.retry_token,
             error_type,
             retry_ready[],
-            ctx
+            pointer_from_objref(ctx)
         )
         if ret != 0
             # NOTE: we're manually creating an AWSError here because calling aws_error
@@ -441,21 +460,23 @@ end
 
 const retry_ready = Ref{Ptr{Cvoid}}(C_NULL)
 
-function c_retry_ready(token::Ptr{aws_retry_token}, error_code::Cint, ctx)
+function c_retry_ready(token, error_code::Cint, ctx_ptr)
+    ctx = unsafe_pointer_to_objref(ctx_ptr)
     if error_code != 0
         ctx.error = CapturedException(aws_error(), Base.backtrace())
         ctx.should_retry = false # don't retry if our retry_schedule failed
         Threads.notify(ctx.completed)
         return
     end
     ctx.verbose >= 2 && @info "retry ready: $(ctx.request.uri.host)"
-    c_on_acquired(C_NULL, 0, token, ctx)
+    c_on_acquired(C_NULL, 0, token, ctx_ptr)
     return
 end
 
 const on_acquired = Ref{Ptr{Cvoid}}(C_NULL)
 
-function c_on_acquired(retry_strategy, error_code, retry_token::Ptr{aws_retry_token}, ctx)
+function c_on_acquired(retry_strategy, error_code, retry_token, ctx_ptr)
+    ctx = unsafe_pointer_to_objref(ctx_ptr)
     if error_code != 0
         ctx.error = CapturedException(aws_error(), Base.backtrace())
         ctx.should_retry = false # don't retry if we failed to get an initial retry_token
@@ -465,7 +486,7 @@ function c_on_acquired(retry_strategy, error_code, retry_token::Ptr{aws_retry_to
     ctx.retry_token = retry_token
     #NOTE: this is threadsafe based on our usage of the default aws connection manager implementation
     ctx.verbose >= 2 && @info "acquired retry token, acquiring connection: $(ctx.request.uri.host)"
-    aws_http_connection_manager_acquire_connection(ctx.client.connection_manager, on_setup[], ctx)
+    aws_http_connection_manager_acquire_connection(ctx.client.connection_manager, on_setup[], ctx_ptr)
 end
 
 get(a...; kw...) = request("GET", a...; kw...)