From 27673cf1d1eb4be3e341bfb642877fdc1568ec9e Mon Sep 17 00:00:00 2001
From: lowit <ea@lowit.ru>
Date: Sun, 1 Sep 2024 02:06:57 +0300
Subject: [PATCH] tarantool: add index for cleansing fiber

---
 hitbox-tarantool/src/init.lua | 26 +++++++++++++++++++++-----
 1 file changed, 21 insertions(+), 5 deletions(-)

diff --git a/hitbox-tarantool/src/init.lua b/hitbox-tarantool/src/init.lua
index 5f48a7d..336a463 100644
--- a/hitbox-tarantool/src/init.lua
+++ b/hitbox-tarantool/src/init.lua
@@ -1,27 +1,43 @@
 local fiber = require("fiber")
 local log = require("log")
 
+local SCAN_INTERVAL = 0.1
+local MAX_TUPLES_FOR_DELETE = 1000
+
 box.cfg({})
 
 local space_name = ...
 
 box.schema.space.create(space_name, { if_not_exists = true })
 box.space[space_name]:create_index("primary", {
-	type = 'HASH',
+	type = "HASH",
 	parts = { { 1, "string" } },
 	if_not_exists = true,
 })
+box.space[space_name]:create_index("by_ttl", {
+	parts = { { 2, "integer" } },
+	if_not_exists = true,
+})
 
 if not _G.__hitbox_cache_fiber then
 	_G.__hitbox_cache_fiber = fiber.create(function()
 		fiber.name("hitbox_cache_fiber")
 		while true do
+			box.ctl.wait_rw()
+
 			local ok, err = pcall(function()
-				for _, t in box.space[space_name]:pairs() do
-					if t[2] <= fiber.time() then
+				local for_del = box.space[space_name].index.by_ttl
+					:pairs({ math.floor(fiber.time()) }, { iterator = "LE" })
+					:take(MAX_TUPLES_FOR_DELETE)
+					:totable()
+
+				box.atomic(function()
+					for _, t in pairs(for_del) do
 						box.space[space_name]:delete(t[1])
 					end
-				end
+				end)
+
+				return true
 			end)
 
 			if not ok then
@@ -29,7 +45,7 @@ if not _G.__hitbox_cache_fiber then
 			end
 
 			fiber.testcancel()
-			fiber.sleep(1)
+			fiber.sleep(SCAN_INTERVAL)
 		end
 	end)
 end