Eventually consistent Master-Master basket

basket.lua

#!/usr/bin/env tarantool
local uuid = require 'uuid'
local fun = require 'fun'
local fiber = require 'fiber'

local instance_name = assert(arg[1], "instance_name is required")

local config = {
	replication = {'127.0.0.1:3301', '127.0.0.1:3302', '127.0.0.1:3303'},
	replication_connect_quorum = 2,
	replication_connect_timeout = 3,
	vinyl_memory = 0,
}
config.listen = 3300 + instance_name:match("(%d+)$")
config.memtx_dir = instance_name
config.wal_dir = instance_name

local function on_conflict(old, new, _, _)
	if not old or not new then -- insert or delete
		return
	end
	local old_items, new_items = old[3], new[3]
	local final_items = {}
	for _, item_id, item_info in fun.chain(old_items, new_items) do
		if not old_items[item_id] or not new_items[item_id] then
			final_items[item_id] = item_info
		elseif old_items[item_id][2] < new_items[item_id][2] then
			final_items[item_id] = new_items[item_id]
		else
			final_items[item_id] = old_items[item_id]
		end
	end
	return box.tuple.new{ old[1], old[2], final_items } -- REPLACE
end

box.ctl.on_schema_init(function()
	box.space._space:on_replace(function(_, space_ddl)
		if space_ddl.name == 'basket' then
			box.on_commit(function()
				local old_trigger = rawget(_G, '\x00basket_before_replace')
				local new_trigger = box.space.basket:before_replace(on_conflict, old_trigger)
				rawset(_G, '\x00basket_before_replace', new_trigger)
			end)
		end
	end)
end)

box.cfg(config)
box.schema.user.grant('guest', 'super', nil, nil, {if_not_exists = true})

box.schema.space.create('basket', {
	format = {
		{ name = 'id',    type = 'uuid'   },
		{ name = 'atime', type = 'number' },
		{ name = 'items', type = '*'      },
	},
	if_not_exists = true,
})

box.space.basket:create_index('primary', { parts = {'id'}, if_not_exists = true })

local basket = {}
_G.basket = basket

--- Helpers
-- basket_get: returns basket or raises error
local function basket_get(basket_id)
	local b = box.space.basket:get(uuid.fromstr(basket_id))
	if not b then
		box.error(
			box.error.TUPLE_NOT_FOUND,
			box.space.basket.index.primary.name,
			box.space.basket.id
		)
	end
	return b:tomap()
end

--- Helper
-- basket_set: updates basket items
local function basket_set(basket, item_id, quantity)
	assert(quantity, "quantity required")
	if basket.items[item_id] then
		basket.items[item_id][1] = math.max(0, basket.items[item_id][1]+quantity)
	else
		basket.items[item_id] = {math.max(0, quantity)}
	end
	basket.items[item_id][2] = fiber.time()
	return box.space.basket:update(basket.id, {
		{ '=', 'atime', fiber.time() },
		{ '=', 'items', basket.items },
	})
end

---
-- API creates new basket
function basket.create()
	return box.space.basket:insert(box.space.basket:frommap {
		id = uuid.new(),
		atime = fiber.time(),
		items = {},
	})
end

---
-- API inserts 1 or more quantity of item into basket
function basket.increase(basket_id, item_id, quantity)
	local b = basket_get(basket_id)
	quantity = quantity or 1
	return basket_set(b, item_id, quantity)
end

---
-- API decrease 1 or more quanties of item in basket
function basket.decrease(basket_id, item_id, quantity)
	local b = basket_get(basket_id)
	quantity = quantity or 1
	return basket_set(b, item_id, -quantity)
end

require 'console'.start() os.exit()