Fix wait_until_synced if you are waiting for your own publish (#28)

* Fix wait_until_synced if you are waiting for your own publish

* fix docstring

* skip stdlibs in downgrade to fix error

Author: Octogonapus

.github/workflows/CI.yml

@@ -103,6 +103,8 @@ jobs:
       - uses: julia-actions/cache@v1
 
       - uses: julia-actions/julia-downgrade-compat@v1
+        with:
+          skip: Dates, Random, Test
 
       - uses: julia-actions/julia-buildpkg@v1
 

Project.toml

@@ -1,6 +1,6 @@
 name = "AWSCRT"
 uuid = "df31ea59-17a4-4ebd-9d69-4f45266dc2c7"
-version = "0.4.1"
+version = "0.4.2"
 
 [deps]
 CountDownLatches = "621fb831-fdad-4fff-93ac-1af7b7ed19e3"

src/ShadowFramework.jl

@@ -232,9 +232,25 @@ end
     wait_until_synced(sf::ShadowFramework)
 
 Blocks until the next time the shadow document is synchronized with the broker.
+
+    !!! warning "Warning: Race Condition"
+        If you are using this function to publish a message and then wait for the following synchronization, you
+        must use the `do` form of [`wait_until_synced`](@ref) instead which accepts a lambda as the first argument.
+        Otherwise, your publish will race the synchronization and there is a chance the synchronization will finish before
+        you begin waiting for it (so you miss the edge and your program hangs).
+"""
+wait_until_synced(sf::ShadowFramework) = wait_until_synced(() -> nothing, sf)
+
+"""
+    wait_until_synced(f::Function, sf::ShadowFramework)
+
+Blocks until the next time the shadow document is synchronized with the broker.
+If you want to wait for a synchronization after a publication that you make, then you must make that publication inside
+the lambda `f`.
 """
-function wait_until_synced(sf::ShadowFramework)
+function wait_until_synced(f::Function, sf::ShadowFramework)
     sf._sync_latch = CountDownLatch(1)
+    f()
     await(sf._sync_latch)
     return nothing
 end

test/shadow_framework_integ_test.jl

@@ -193,8 +193,9 @@ shadow_types = parallel ? [:named] : [:named, :unnamed]
         try
             fetch(publish(sc, "/delete", "", AWS_MQTT_QOS_AT_LEAST_ONCE)[1]) # ensure the shadow is deleted just in case of a prior broken test
 
-            fetch(subscribe(sf)[1]) # subscribe and trigger the initial update, which will fail because there is no shadow
-            wait_until_synced(sf)
+            wait_until_synced(sf) do
+                fetch(subscribe(sf)[1]) # subscribe and trigger the initial update, which will fail because there is no shadow
+            end
             sleep(3) # we need to make sure the local shadow won't get modified. no better way than to just wait a bit in case something modifies it.
             @test collect(keys(doc)) == ["version", "foo"] # we should have the version and the initial foo key we set
             @test doc["foo"] == 1 # should be unchanged from our initial state
@@ -241,8 +242,9 @@ shadow_types = parallel ? [:named] : [:named, :unnamed]
 
             @info "subscribing in band shadow"
             values_post_update = []
-            fetch(subscribe(sf)[1]) # subscribe and trigger the initial update
-            wait_until_synced(sf)
+            wait_until_synced(sf) do
+                fetch(subscribe(sf)[1]) # subscribe and trigger the initial update
+            end
             wait_for(() -> length(values_post_update) >= 1) # wait for the update to finish since it requires multiple messages
             # The initial update should have pulled in that desired state
             @test doc["foo"] == 2
@@ -425,10 +427,11 @@ end
 
         try
             @info "subscribing"
-            fetch(subscribe(sf)[1])
             # wait for the first publish to finish, otherwise we will race it with our next update, which could arrive
             # first and break this test
-            wait_until_synced(sf)
+            wait_until_synced(sf) do
+                fetch(subscribe(sf)[1])
+            end
 
             # publish a /update. this should be accepted. the local shadow should be updated.
             # an /update should be published with the new reported state.
@@ -516,10 +519,11 @@ end
 
         try
             @info "subscribing"
-            fetch(subscribe(sf)[1])
             # wait for the first publish to finish, otherwise we will race it with our next update, which could arrive
             # first and break this test
-            wait_until_synced(sf)
+            wait_until_synced(sf) do
+                fetch(subscribe(sf)[1])
+            end
 
             # publish an /update which adds bar=1. this should be rejected because bar is not present in the struct.
             # the local shadow should not be updated. an /update should not be published.