fix(clustering/sync): fix delta not using absolute TTL

kong/clustering/services/sync/hooks.lua

@@ -9,6 +9,7 @@ local EMPTY = require("kong.tools.table").EMPTY
 local ipairs = ipairs
 local ngx_log = ngx.log
 local ngx_ERR = ngx.ERR
+local ngx_WARN = ngx.WARN
 local ngx_DEBUG = ngx.DEBUG
 
 
@@ -76,6 +77,19 @@ end
 
 
 function _M:entity_delta_writer(entity, name, options, ws_id, is_delete)
+  if not is_delete and entity and entity.ttl then
+    -- Replace relative TTL value to absolute TTL value
+    local exported_entity = self.strategy:export_entity(name, entity, options)
+
+    if exported_entity and exported_entity.ttl then
+      ngx_log(ngx_DEBUG, "[kong.sync.v2] Update TTL from relative value to absolute value ", exported_entity.ttl, ".")
+      entity.ttl = exported_entity.ttl
+
+    else
+      ngx_log(ngx_WARN, "[kong.sync.v2] Cannot update TTL of entity (", name, ") to absolute value.")
+    end
+  end
+
   local res, err = self.strategy:insert_delta()
   if not res then
     self.strategy:cancel_txn()

kong/clustering/services/sync/strategies/postgres.lua

@@ -1,6 +1,7 @@
 local _M = {}
 local _MT = { __index = _M }
 
+local kong_table = require("kong.tools.table")
 
 local sub = string.sub
 local fmt = string.format
@@ -15,6 +16,7 @@ local VERSION_FMT = VER_PREFIX .. "%028x"
 
 function _M.new(db)
   local self = {
+    db = db,
     connector = db.connector,
   }
 
@@ -64,6 +66,13 @@ function _M:is_valid_version(str)
 end
 
 
+function _M:export_entity(name, entity, options)
+  local options = kong_table.cycle_aware_deep_copy(options, true)
+  options["export"] = true
+  return self.db[name]:select(entity, options)
+end
+
+
 function _M:begin_txn()
   return self.connector:query("BEGIN;")
 end

kong/db/strategies/postgres/init.lua

@@ -570,8 +570,14 @@ end
 
 function _mt:select(primary_key, options)
   local statement_name = "select"
-  if self.schema.ttl and options and options.skip_ttl then
-    statement_name = "select_skip_ttl"
+  if options then
+    -- If we do select for export, there is no need to skip ttl.
+    if options.export then
+      statement_name = "select_for_export"
+
+    elseif self.schema.ttl and options.skip_ttl then
+      statement_name = "select_skip_ttl"
+    end
   end
 
   local res, err = execute(self, statement_name, self.collapse(primary_key), options)
@@ -1248,6 +1254,22 @@ function _M.new(connector, schema, errors)
     }
   })
 
+  add_statement_for_export("select", {
+    operation = "read",
+    expr = select_expressions,
+    argn = primary_key_names,
+    argv = primary_key_args,
+    code = {
+      "SELECT ",  select_expressions, "\n",
+      "  FROM ",  table_name_escaped, "\n",
+      where_clause(
+      " WHERE ", "(" .. pk_escaped .. ") = (" .. primary_key_placeholders .. ")",
+                 ttl_select_where,
+                 ws_id_select_where),
+      " LIMIT 1;"
+    }
+  })
+
   add_statement_for_export("page_first", {
     operation = "read",
     argn = { LIMIT },