Merge pull request #158 from hamishforbes/master

Fixes for metadata transaction abort

lib/ledge.lua

@@ -155,6 +155,7 @@ local function create_redis_connection()
     if not rc then
         return nil, err
     end
+
     return rc:connect()
 end
 _M.create_redis_connection = create_redis_connection
@@ -170,18 +171,25 @@ local function create_redis_slave_connection()
     if not rc then
         return nil, err
     end
+
     return rc:connect()
 end
 _M.create_redis_slave_connection = create_redis_slave_connection
 
 
 local function close_redis_connection(redis)
+    if not next(redis) then
+        -- Possible for this to be called before we've created a redis conn
+        -- Ensure we actually have a resty-redis instance to close
+        return nil
+    end
+
     local rc, err = redis_connector.new(config.redis_connector_params)
     if not rc then
         return nil, err
     end
-    return rc:set_keepalive(redis)
 
+    return rc:set_keepalive(redis)
 end
 _M.close_redis_connection = close_redis_connection
 

lib/ledge/handler.lua

@@ -568,7 +568,8 @@ local function revalidate_in_background(self, update_revalidation_data)
 
         local ttl, err = redis:ttl(key_chain.reval_params)
         if not ttl or ttl == ngx_null or ttl < 0 then
-            ngx_log(ngx_ERR,
+            if err then ngx_log(ngx_ERR, err) end
+            ngx_log(ngx_INFO,
                 "Could not determine expiry for revalidation params. " ..
                 "Will fallback to 3600 seconds."
             )
@@ -577,27 +578,38 @@ local function revalidate_in_background(self, update_revalidation_data)
         end
 
         -- Delete and update reval request headers
-        redis:multi()
-
-        redis:del(key_chain.reval_params)
-        redis:hmset(key_chain.reval_params, reval_params)
-        redis:expire(key_chain.reval_params, ttl)
-
-        redis:del(key_chain.reval_req_headers)
-        redis:hmset(key_chain.reval_req_headers, reval_headers)
-        redis:expire(key_chain.reval_req_headers, ttl)
+        local _, e
+        _, e = redis:multi()
+        if e then ngx_log(ngx_ERR, e) end
+        _, e = redis:del(key_chain.reval_params)
+        if e then ngx_log(ngx_ERR, e) end
+        _, e = redis:hmset(key_chain.reval_params, reval_params)
+        if e then ngx_log(ngx_ERR, e) end
+        _, e = redis:expire(key_chain.reval_params, ttl)
+        if e then ngx_log(ngx_ERR, e) end
+
+        _, e = redis:del(key_chain.reval_req_headers)
+        if e then ngx_log(ngx_ERR, e) end
+        _, e = redis:hmset(key_chain.reval_req_headers, reval_headers)
+        if e then ngx_log(ngx_ERR, e) end
+        _, e = redis:expire(key_chain.reval_req_headers, ttl)
+        if e then ngx_log(ngx_ERR, e) end
 
         local res, err = redis:exec()
-        if not res then
+        if not res or res == ngx_null then
             ngx_log(ngx_ERR, "Could not update revalidation params: ", err)
         end
     end
 
     local uri, err = redis:hget(key_chain.main, "uri")
     if not uri or uri == ngx_null then
-        ngx_log(ngx_ERR,
-            "Cache key has no 'uri' field, aborting revalidation"
-        )
+        if err then
+            ngx_log(ngx_ERR, "Failed to get main key while revalidating: ", err)
+        else
+            ngx_log(ngx_WARN,
+                "Cache key has no 'uri' field, aborting revalidation"
+            )
+        end
         return nil
     end
 
@@ -665,7 +677,7 @@ local function save_to_cache(self, res)
     -- and the size.
     local previous_entity_id = self:entity_id(key_chain)
 
-    local previous_entity_size, err
+    local previous_entity_size, err, gc_job_spec
     if previous_entity_id then
         previous_entity_size, err = redis:hget(key_chain.main, "size")
         if previous_entity_size == ngx_null then
@@ -674,14 +686,9 @@ local function save_to_cache(self, res)
                 ngx_log(ngx_ERR, err)
             end
         end
-    end
-
-    -- Start the transaction
-    local ok, err = redis:multi()
-    if not ok then ngx_log(ngx_ERR, err) end
 
-    if previous_entity_id then
-        put_background_job(
+        -- Define GC job here, used later if required
+        gc_job_spec = {
             "ledge_gc",
             "ledge.jobs.collect_entity",
             {
@@ -697,13 +704,18 @@ local function save_to_cache(self, res)
                 tags = { "collect_entity" },
                 priority = 10,
             }
-        )
+        }
+    end
 
+    -- Start the transaction
+    local ok, err = redis:multi()
+    if not ok then ngx_log(ngx_ERR, err) end
+
+    if previous_entity_id then
         local ok, err = redis:srem(key_chain.entities, previous_entity_id)
         if not ok then ngx_log(ngx_ERR, err) end
     end
 
-
     res.uri = req_full_uri()
 
     local keep_cache_for = self.config.keep_cache_for
@@ -749,7 +761,20 @@ local function save_to_cache(self, res)
 
             ok, e = redis:exec()
             if not ok or ok == ngx_null then
-                ngx_log(ngx_ERR, "failed to complete transaction: ", e)
+                if e then
+                    ngx_log(ngx_ERR, "failed to complete transaction: ", e)
+                else
+                    -- Transaction likely failed due to watch on main key
+                    -- Tell storage to clean up too
+                    ok, e = storage:delete(res.entity_id)
+                    if e then
+                        ngx_log(ngx_ERR, "failed to cleanup storage: ", e)
+                    end
+                end
+            elseif previous_entity_id then
+                -- Everything has completed and we have an old entity
+                -- Schedule GC to clean it up
+                put_background_job(unpack(gc_job_spec))
             end
         end
 
@@ -780,6 +805,10 @@ local function save_to_cache(self, res)
         local ok, e = redis:exec()
         if not ok or ok == ngx_null then
             ngx_log(ngx_ERR, "failed to complete transaction: ", e)
+        elseif previous_entity_id then
+            -- Everything has completed and we have an old entity
+            -- Schedule GC to clean it up
+            put_background_job(unpack(gc_job_spec))
         end
     end
 end

t/02-integration/cache.t

@@ -11,6 +11,8 @@ $ENV{TEST_COVERAGE} ||= 0;
 our $HttpConfig = qq{
 lua_package_path "./lib/?.lua;../lua-resty-redis-connector/lib/?.lua;../lua-resty-qless/lib/?.lua;../lua-resty-http/lib/?.lua;../lua-ffi-zlib/lib/?.lua;;";
 
+lua_shared_dict ledge_test 1m;
+
 init_by_lua_block {
     if $ENV{TEST_COVERAGE} == 1 then
         require("luacov.runner").init()
@@ -705,10 +707,104 @@ Numkeys: 5
 [error]
 
 
-=== TEST 15c: Partial entry misses
+=== TEST 16: Prime a resource into cache
 --- http_config eval: $::HttpConfig
 --- config
-location /cache_15_prx {
+location /cache_16_prx {
+    rewrite ^(.*)_prx$ $1 break;
+    content_by_lua_block {
+        require("ledge").create_handler():run()
+    }
+}
+location /cache_16 {
+    content_by_lua_block {
+        ngx.header["Cache-Control"] = "max-age=60"
+        ngx.say("TEST 16")
+    }
+}
+--- request
+GET /cache_16_prx
+--- response_headers_like
+X-Cache: MISS from .*
+--- response_body
+TEST 16
+--- no_error_log
+[error]
+
+=== TEST 16b: Modified main key aborts transaction and cleans up entity
+--- http_config eval: $::HttpConfig
+--- config
+location /cache_16_check {
+    content_by_lua_block {
+        local entity_id = ngx.shared.ledge_test:get("entity_id")
+        local redis = require("ledge").create_storage_connection()
+        local ok, err = redis:exists(entity_id)
+        ngx.print(ok, " ", err)
+    }
+}
+location /cache_16_prx {
+    rewrite ^(.*)_prx$ $1 break;
+    content_by_lua_block {
+        local handler = require("ledge").create_handler()
+        handler:bind("before_serve", function(res)
+            -- Create a new connection
+            local redis = require("ledge").create_redis_connection()
+            -- Set a new key on the main key
+            redis:hset(handler:cache_key_chain().main, "foo", "bar")
+
+            ngx.shared.ledge_test:set("entity_id", res.entity_id)
+        end)
+
+        handler:run()
+    }
+}
+location /cache_16 {
+    content_by_lua_block {
+        ngx.header["Cache-Control"] = "max-age=60"
+        ngx.print("TEST 16b")
+    }
+}
+--- request eval
+["GET /cache_16_prx", "GET /cache_16_check"]
+--- more_headers
+Cache-Control: no-cache
+--- response_headers_like eval
+["X-Cache: MISS from .*", ""]
+--- response_body eval
+["TEST 16b", "false nil"]
+--- wait: 3
+--- no_error_log
+[error]
+
+=== TEST 16c: Modified main key aborts transaction - HIT
+--- http_config eval: $::HttpConfig
+--- config
+location /cache_16_prx {
+    rewrite ^(.*)_prx$ $1 break;
+    content_by_lua_block {
+        require("ledge").create_handler():run()
+    }
+}
+location /cache_16 {
+    content_by_lua_block {
+        ngx.header["Cache-Control"] = "max-age=60"
+        ngx.say("TEST 16b")
+    }
+}
+--- request
+GET /cache_16_prx
+--- response_headers_like
+X-Cache: HIT from .*
+--- response_body
+TEST 16
+--- no_error_log
+[error]
+
+
+=== TEST 16d: Partial entry misses
+--- http_config eval: $::HttpConfig
+--- config
+location /cache_16_prx {
     rewrite ^(.*)_prx$ $1 break;
     content_by_lua_block {
         local handler = require("ledge").create_handler()
@@ -721,17 +817,17 @@ location /cache_15_prx {
         handler:run()
     }
 }
-location /cache_15 {
+location /cache_16 {
     content_by_lua_block {
         ngx.header["Cache-Control"] = "max-age=60"
-        ngx.say("TEST 15c")
+        ngx.say("TEST 16d")
     }
 }
 --- request
-GET /cache_15_prx
+GET /cache_16_prx
 --- response_headers_like
 X-Cache: MISS from .*
 --- response_body
-TEST 15c
+TEST 16d
 --- no_error_log
 [error]