diff --git a/guides/configuration.md b/guides/configuration.md index 5a124e1..68c0c9e 100644 --- a/guides/configuration.md +++ b/guides/configuration.md @@ -845,6 +845,21 @@ to the console, which is off until asked for. It has no environment variable; set it in `sys.config`. +### Node-local KV + +`game.kv` is a separate, in-memory tier for state that has to outlive a zone at +tick rate - see [Lua API](lua-api.md#node-local-kv). It is not part of the +storage subsystem and is not switched off with it: it touches no database and +serves no route. + +| Key | Default | Description | +|-----|---------|-------------| +| `kv_ttl_seconds` | `3600` | How long an untouched `game.kv` entry lives. Every write refreshes it, so state in active use never ages out | +| `kv_max_keys` | `100000` | Cap on distinct keys per node. Past it a new key is refused and asobi logs `kv_full`; an existing key stays writable | + +Both are per node and take effect on the next write, so a running node can be +retuned without a restart. + ## Vote templates Reusable vote configurations, merged with the per-vote config from your game diff --git a/guides/large-worlds.md b/guides/large-worlds.md index 1a24a67..cb73721 100644 --- a/guides/large-worlds.md +++ b/guides/large-worlds.md @@ -126,7 +126,9 @@ So a persistent world gets the zone back as it was. A non-persistent one gets back only "what was true here" - the `zone_state` your `on_zone_loaded` or `zone_tick` last left - and re-spawns the rest. That is enough for content that is a pure function of its coordinates; it is not enough for a damaged NPC or a -half-looted container, and those need `persistent = true`. +half-looted container, and those need `persistent = true` - or a tier that is +not zone-scoped at all, which is +[State that outlives a zone](#state-that-outlives-a-zone). The in-memory `zone_state` set is capped at `max_active_zones` entries and dies with the world. Past the cap a parked zone comes back blank, and asobi @@ -237,6 +239,33 @@ provider. It starts automatically when the game returns a terrain provider, and caches each chunk after first load. It is per node, like every other cache - see [Clustering](clustering.md#what-is-per-node). +## State that outlives a zone + +An entity that crosses the map - a freighter on a trade route, a patrol, a +convoy - has state that is neither recomputable nor zone-scoped. Position can be +made a pure function of a clock and needs no storage at all; hull points, "this +one is already dead", "it has shed 4 of its 10 containers" cannot. + +Neither place a zone script already has works for it. The zone's entity map goes +with the zone. A table in the zone's Lua VM is one copy per zone and goes with +the VM. And an entity crossing a seam is legitimately held by two zones at once +until it is `rehome_margin` past the boundary, so both tick and both can write +to it - two schedulers, no ordering between them. + +Two tiers, and the difference that matters is what survives a node restart: + +| | `game.kv` | `game.storage.update` | +|---|---|---| +| Where | Node memory, scoped to this world | Database, global | +| Survives a restart | **No** | Yes | +| Cost per write | A message to one process | Two database round trips | +| Concurrent writers | Atomic merge | Atomic merge, version-checked | +| Use it for | Per-hit damage, per-tick accumulation | Per-kill, per-threshold | + +Both take the same order-independent [merge +operators](lua-api.md#merge-operators), so two zones can each apply what they +saw with no lock and no lost write. + ## Zone lifecycle callbacks A world script can react to zones loading and unloading. Both callbacks are diff --git a/guides/lua-api.md b/guides/lua-api.md index 6dc1ec2..ad06641 100644 --- a/guides/lua-api.md +++ b/guides/lua-api.md @@ -12,7 +12,7 @@ into a sandboxed VM with **no `game` table at all** - see There are two return conventions and you have to know which one you are looking at. -The persistence-style calls - `economy.*`, `storage.*`, +The persistence-style calls - `economy.*`, `storage.*`, `kv.*`, `leaderboard.top/rank/around`, `terrain.get_chunk`, `notify`, `notify_many` - return a wrapped table. Index into `.ok`: @@ -219,8 +219,10 @@ to you. ```lua game.storage.get(collection, key) game.storage.set(collection, key, value) +game.storage.update(collection, key, ops) game.storage.player_get(player_id, collection, key) game.storage.player_set(player_id, collection, key, value) +game.storage.player_update(player_id, collection, key, ops) ``` Two distinct namespaces, not one with an optional owner. `get`/`set` write @@ -233,6 +235,104 @@ overwrite what `game.storage.set` wrote. That makes it the right place for server-authoritative configuration and the wrong place for anything a client needs to fetch directly. +### set loses concurrent writes; update does not + +`set` reads the row and then writes it, and nothing holds the two together. Two +writers racing on one key therefore lose a write, silently. That is fine for +"save the sector's config at boot" and wrong for anything two zones can touch. + +`update` applies [merge operators](#merge-operators) server-side, conditional on +the version it read. A racing writer's update matches no rows, and this re-reads +and re-applies - which is correct rather than merely safe, because the operators +are order-independent. After five attempts on a genuinely hot key it answers +`{ error = "storage_conflict" }` instead of spinning a zone that has a +millisecond budget. + +```lua +local r = game.storage.update("ships", ship_id, { + hull = { incr = -40 }, + kills = { incr = 1 }, + dead = { latch = true }, +}) +if r.ok.hull <= 0 then ... end +``` + +It is still a database round trip inside a callback, so it suits a per-kill or +per-threshold write. For a per-hit one, use `game.kv`. + +## Node-local KV + +```lua +game.kv.get(collection, key) +game.kv.set(collection, key, value) +game.kv.set(collection, key, value, ttl_seconds) +game.kv.merge(collection, key, ops) +game.kv.merge(collection, key, ops, ttl_seconds) +game.kv.delete(collection, key) +``` + +State that has to outlive the zone holding it, at tick rate. An entity that +crosses the map - a freighter on a trade route, a patrol - has state that is +neither recomputable nor zone-scoped: hull points, "this one is already dead", +"it has shed 4 of its 10 containers". The zone's entity map goes with the zone +and a table in the zone's VM goes with the VM, so neither works. + +- **In memory, on one node.** Lost on restart, never replicated. A world lives + on one node, so this is per world in practice - but say it out loud in your + design. Anything that must survive a deploy belongs in `game.storage`. +- **Scoped to the calling world or match.** Two worlds running the same mode + script cannot collide on the same collection and key. +- **Reads are free.** `get` reads shared memory from the calling process, with + no message round trip, so a tick can read state without paying for it. +- **Merges are atomic.** Every write is serialised, so a read-modify-write + cannot interleave, and `merge` answers with the merged value - the caller + usually needs to know whether that was the hit that killed it. +- **TTL'd.** Every write refreshes the entry's expiry, so state in active use + never ages out; an entry nothing has touched for an hour is swept. Pass + `ttl_seconds` for a different lifetime. + +```lua +function handle_effects(effects, entities) + for _, e in ipairs(effects) do + local r = game.kv.merge("ships", e.target, { + hull = { incr = -e.damage }, + dead = { latch = e.damage >= 999 }, + }) + if r.ok.hull <= 0 then game.zone.despawn(e.target) end + end + return entities +end +``` + +`set` takes a table, not a scalar: everything here merges field by field. + +### Merge operators + +`game.kv.merge` and `game.storage.update` take the same shape - a table of +field names to a table naming exactly one operator: + +| Operator | Does | Order-independent | +|---|---|---| +| `set` | Replaces the field | **No** - last writer wins | +| `set_if_absent` | Writes only if the field is missing | Yes | +| `incr` | Adds a number (negative to subtract) | Yes | +| `min` | Keeps the lower of the two | Yes | +| `max` | Keeps the higher of the two | Yes | +| `latch` | Boolean OR: once true, stays true | Yes | + +Order-independence is the point. An entity crossing a zone boundary is +legitimately held by two zones at once until it is `rehome_margin` past the +seam, and both tick - two schedulers, no ordering between them. Every operator +above except `set` gives the same answer whichever lands first, so neither zone +needs a lock, a version, or a re-read. `set` is honest about being +last-writer-wins; put it on a field only one writer owns. + +Anything else is an error naming the field: an unknown operator, a bare value +(`{ hull = 40 }` rather than `{ hull = { set = 40 } }`), two operators in one +field, or an operator applied to the wrong type. A bare value is deliberately +*not* treated as `set` - it would read exactly like a working merge and lose +writes. + ## Chat ```lua diff --git a/src/asobi_kv.erl b/src/asobi_kv.erl new file mode 100644 index 0000000..e225cc4 --- /dev/null +++ b/src/asobi_kv.erl @@ -0,0 +1,255 @@ +-module(asobi_kv). +-behaviour(gen_server). + +-moduledoc """ +A node-local, in-memory key-value store for state that has to outlive the zone +holding it, at tick rate. + +## Why this exists + +A world with entities that cross the map - a freighter on a trade route, a +patrol, a convoy - needs per-entity state that survives the zone. Position can +be a pure function of a clock and needs no storage; hull points, "this one is +already dead", "it has shed 4 of its 10 containers" cannot be recomputed from +anything. + +Neither place a zone script already has works for that. The zone's entity map +goes with the zone. A table in the zone's Lua VM is one copy per zone and goes +with the VM. And `game.storage` - which has exactly the right semantics - is a +synchronous database read-modify-write with no atomic update, executed inside a +callback budgeted in milliseconds. Fine for "save the sector's config at boot"; +not usable for "this ship just took 40 damage" (widgrensit/asobi#572). + +## What it promises, and what it does not + +- **In memory, on one node.** Lost on restart, and never replicated: a world + lives entirely on one node, so this is per world in practice. Say it out + loud in your design - anything that must survive a deploy belongs in + `game.storage`. +- **Atomic merges.** Every write goes through this process, so a + read-modify-write cannot interleave. Combined with the commutative operators + in `m:asobi_merge_ops`, two zones holding the same entity across a seam can + both apply what they saw without a lock, a version or a lost write. +- **Reads do not touch this process.** `get/3` is an ETS read from the calling + process, so a zone tick reading state pays no message round trip. +- **Scoped per world or match.** The scope comes from the calling VM, so two + worlds running the same mode script cannot collide on the same key. +- **TTL'd.** Every write refreshes the entry's expiry, so state in active use + never ages out; state nothing has touched for `kv_ttl_seconds` (an hour by + default) is swept. That is what keeps a long-lived node from accumulating + one entry per entity that ever existed. +""". + +-export([start_link/0]). +-export([get/3, set/4, set/5, merge/4, merge/5, delete/3]). +-export([init/1, handle_call/3, handle_cast/2, handle_info/2]). + +-include_lib("kernel/include/logger.hrl"). + +-define(TABLE, asobi_kv). +-define(SWEEP_INTERVAL, 60_000). +-define(DEFAULT_TTL_SECONDS, 3600). +-define(DEFAULT_MAX_KEYS, 100_000). +-define(CALL_TIMEOUT, 5_000). + +-type scope() :: binary(). + +%% --- Public API --- + +-spec start_link() -> gen_server:start_ret(). +start_link() -> + gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). + +-doc """ +Read a key. An ETS read in the calling process - no message to this server. + +An entry past its expiry reads as `not_found` whether or not the sweep has got +to it yet, so a caller never sees a value the TTL says is gone. +""". +-spec get(scope(), binary(), binary()) -> {ok, map()} | not_found. +get(Scope, Collection, Key) -> + try ets:lookup(?TABLE, {Scope, Collection, Key}) of + [{_K, Value, ExpiresAt}] when is_map(Value) -> + case expired(ExpiresAt) of + true -> not_found; + false -> {ok, Value} + end; + _ -> + not_found + catch + %% The table is gone: this server is restarting, or it never started + %% (a release that does not run asobi_sup's engine children). A miss is + %% the honest answer for a store that promises nothing across a + %% restart anyway. + error:badarg -> not_found + end. + +-spec set(scope(), binary(), binary(), map()) -> ok | {error, term()}. +set(Scope, Collection, Key, Value) -> + set(Scope, Collection, Key, Value, default_ttl()). + +-spec set(scope(), binary(), binary(), map(), pos_integer()) -> ok | {error, term()}. +set(Scope, Collection, Key, Value, TtlSeconds) when is_map(Value) -> + call_unit({set, {Scope, Collection, Key}, Value, TtlSeconds}). + +-doc """ +Apply `m:asobi_merge_ops` to the value at this key and return the result, +creating the entry from `#{}` if it is absent. + +A call, not a cast: a caller merging damage into a shared entity almost always +needs to know what the value became - whether that was the hit that killed it - +and a cast cannot answer that. +""". +-spec merge(scope(), binary(), binary(), map()) -> {ok, map()} | {error, term()}. +merge(Scope, Collection, Key, Ops) -> + merge(Scope, Collection, Key, Ops, default_ttl()). + +-spec merge(scope(), binary(), binary(), map(), pos_integer()) -> {ok, map()} | {error, term()}. +merge(Scope, Collection, Key, Ops, TtlSeconds) when is_map(Ops) -> + call_map({merge, {Scope, Collection, Key}, Ops, TtlSeconds}). + +-spec delete(scope(), binary(), binary()) -> ok | {error, term()}. +delete(Scope, Collection, Key) -> + call_unit({delete, {Scope, Collection, Key}}). + +%% --- gen_server callbacks --- + +-spec init([]) -> {ok, map()}. +init([]) -> + %% Public and read_concurrency: get/3 reads it straight from the caller. + %% Writes all come through this process, which is what makes a merge + %% atomic without a lock. + _ = ets:new(?TABLE, [set, public, named_table, {read_concurrency, true}]), + schedule_sweep(), + {ok, #{}}. + +-spec handle_call(term(), gen_server:from(), map()) -> {reply, term(), map()}. +handle_call({set, Key, Value, TtlSeconds}, _From, State) -> + case admit(Key) of + ok -> + ets:insert(?TABLE, {Key, Value, expires_at(TtlSeconds)}), + {reply, ok, State}; + {error, _} = Err -> + {reply, Err, State} + end; +handle_call({merge, Key, Ops, TtlSeconds}, _From, State) when is_map(Ops) -> + case admit(Key) of + ok -> + case asobi_merge_ops:apply_ops(current(Key), Ops) of + {ok, Merged} -> + ets:insert(?TABLE, {Key, Merged, expires_at(TtlSeconds)}), + {reply, {ok, Merged}, State}; + {error, _} = Err -> + {reply, Err, State} + end; + {error, _} = Err -> + {reply, Err, State} + end; +handle_call({delete, Key}, _From, State) -> + ets:delete(?TABLE, Key), + {reply, ok, State}; +handle_call(_Request, _From, State) -> + {reply, {error, unknown_request}, State}. + +-spec handle_cast(term(), map()) -> {noreply, map()}. +handle_cast(_Msg, State) -> + {noreply, State}. + +-spec handle_info(term(), map()) -> {noreply, map()}. +handle_info(sweep, State) -> + Now = now_ms(), + _ = ets:select_delete(?TABLE, [{{'_', '_', '$1'}, [{'<', '$1', Now}], [true]}]), + schedule_sweep(), + {noreply, State}; +handle_info(_Info, State) -> + {noreply, State}. + +%% --- Internal --- + +%% A cap, because nothing here has an owner that outlives it: an entry is +%% dropped by its TTL or by nothing at all, and a game that mints a key per +%% projectile would otherwise grow this table until the node died. An existing +%% key is always writable - the cap bounds distinct keys, it does not stop a +%% game updating what it already has. +%% Takes the key off the call payload, which gen_server types as term(). +-spec admit(term()) -> ok | {error, kv_full}. +admit(Key) -> + Max = max_keys(), + case ets:info(?TABLE, size) >= Max andalso not ets:member(?TABLE, Key) of + false -> + ok; + true -> + ?LOG_WARNING(#{ + event => kv_full, + max_keys => Max, + msg => ~"game.kv is at kv_max_keys; new keys are refused until entries expire" + }), + {error, kv_full} + end. + +-spec current(term()) -> map(). +current(Key) -> + case ets:lookup(?TABLE, Key) of + [{_K, Value, ExpiresAt}] when is_map(Value) -> + case expired(ExpiresAt) of + true -> #{}; + false -> Value + end; + _ -> + #{} + end. + +%% Two narrowings of the same call rather than one `term()`: a gen_server reply +%% is untyped, and every caller here has a shape it must answer with. +-spec call_unit(term()) -> ok | {error, term()}. +call_unit(Msg) -> + case call(Msg) of + ok -> ok; + {error, Reason} -> {error, Reason}; + Other -> {error, {unexpected_reply, Other}} + end. + +-spec call_map(term()) -> {ok, map()} | {error, term()}. +call_map(Msg) -> + case call(Msg) of + {ok, Value} when is_map(Value) -> {ok, Value}; + {error, Reason} -> {error, Reason}; + Other -> {error, {unexpected_reply, Other}} + end. + +-spec call(term()) -> term(). +call(Msg) -> + try + gen_server:call(?MODULE, Msg, ?CALL_TIMEOUT) + catch + exit:{noproc, _} -> {error, kv_unavailable}; + exit:{timeout, _} -> {error, kv_timeout} + end. + +-spec max_keys() -> pos_integer(). +max_keys() -> + case application:get_env(asobi, kv_max_keys, ?DEFAULT_MAX_KEYS) of + N when is_integer(N), N > 0 -> N; + _ -> ?DEFAULT_MAX_KEYS + end. + +-spec default_ttl() -> pos_integer(). +default_ttl() -> + case application:get_env(asobi, kv_ttl_seconds, ?DEFAULT_TTL_SECONDS) of + N when is_integer(N), N > 0 -> N; + _ -> ?DEFAULT_TTL_SECONDS + end. + +expires_at(TtlSeconds) when is_integer(TtlSeconds), TtlSeconds > 0 -> + now_ms() + TtlSeconds * 1000; +expires_at(_) -> + now_ms() + default_ttl() * 1000. + +expired(ExpiresAt) when is_integer(ExpiresAt) -> ExpiresAt < now_ms(); +expired(_) -> false. + +now_ms() -> + erlang:monotonic_time(millisecond). + +schedule_sweep() -> + erlang:send_after(?SWEEP_INTERVAL, self(), sweep). diff --git a/src/asobi_merge_ops.erl b/src/asobi_merge_ops.erl new file mode 100644 index 0000000..a0e9cc9 --- /dev/null +++ b/src/asobi_merge_ops.erl @@ -0,0 +1,110 @@ +-module(asobi_merge_ops). + +-moduledoc """ +The operator vocabulary shared by `game.kv.merge` and `game.storage.update`. + +A merge is a map of field names to a single-key operator table: + +```lua +{ + hull = { min = 40 }, -- lowest wins + dead = { latch = true }, -- once true, stays true + shed = { max = 4 }, -- highest wins + kills = { incr = 1 }, -- add + name = { set = "Kestrel" } -- last writer wins +} +``` + +Operators rather than a read-modify-write is the whole point. Every operator +here except `set` is commutative and idempotent-under-reordering, so two zones +that both hold the same entity - which the engine allows on purpose, until it +is `rehome_margin` past the boundary - can each apply what they saw with no +lock, no version and no lost write. Two schedulers, no ordering, same answer. + +`set` is the exception and is honest about it: it is last-writer-wins, so it +belongs on a field only one writer owns. + +An unknown operator, a field whose value is not a single-key operator table, +or an operator applied to the wrong type is an error naming the field. Silently +treating a malformed entry as `set` is the failure mode this exists to avoid: +it would look exactly like a working merge and lose writes. +""". + +-export([apply_ops/2, operators/0]). + +-export_type([ops/0, operator/0]). + +-type operator() :: set | set_if_absent | incr | min | max | latch. +-type ops() :: #{binary() => map()}. + +-doc "Every operator name, as it is written in Lua.". +-spec operators() -> [binary(), ...]. +operators() -> + [~"set", ~"set_if_absent", ~"incr", ~"min", ~"max", ~"latch"]. + +-doc """ +Apply `Ops` to `Value`, returning the merged map or the first field that did +not make sense. +""". +-spec apply_ops(map(), map()) -> {ok, map()} | {error, binary()}. +apply_ops(Value, Ops) when is_map(Value), is_map(Ops) -> + apply_each(maps:to_list(Ops), Value); +apply_ops(_Value, _Ops) -> + {error, ~"merge requires a table of field operators"}. + +%% Explicit recursion: see docs/eqwalizer-idioms.md. +-spec apply_each([{term(), term()}], map()) -> {ok, map()} | {error, binary()}. +apply_each([], Value) -> + {ok, Value}; +apply_each([{Field, Op} | Rest], Value) when is_binary(Field) -> + case apply_one(Field, Op, Value) of + {ok, Value1} -> apply_each(Rest, Value1); + {error, _} = Err -> Err + end; +apply_each([{Field, _Op} | _Rest], _Value) -> + {error, <<"merge field is not a name: ", (format_term(Field))/binary>>}. + +-spec apply_one(binary(), term(), map()) -> {ok, map()} | {error, binary()}. +apply_one(Field, Op, Value) when is_map(Op), map_size(Op) =:= 1 -> + [{OpName, Arg}] = maps:to_list(Op), + case lists:member(OpName, operators()) of + true -> operate(OpName, Field, Arg, Value); + false -> {error, <<"unknown merge operator on ", Field/binary>>} + end; +apply_one(Field, _Op, _Value) -> + {error, + <<"merge field ", Field/binary, + " must be a table naming exactly one operator, e.g. { min = 40 }">>}. + +-spec operate(term(), binary(), term(), map()) -> {ok, map()} | {error, binary()}. +operate(~"set", Field, Arg, Value) -> + {ok, Value#{Field => Arg}}; +operate(~"set_if_absent", Field, Arg, Value) -> + case maps:is_key(Field, Value) of + true -> {ok, Value}; + false -> {ok, Value#{Field => Arg}} + end; +operate(~"incr", Field, Arg, Value) when is_number(Arg) -> + case maps:get(Field, Value, 0) of + Current when is_number(Current) -> {ok, Value#{Field => Current + Arg}}; + _ -> {error, <<"incr on a non-numeric field: ", Field/binary>>} + end; +operate(~"min", Field, Arg, Value) when is_number(Arg) -> + case maps:get(Field, Value, Arg) of + Current when is_number(Current) -> {ok, Value#{Field => min(Current, Arg)}}; + _ -> {error, <<"min on a non-numeric field: ", Field/binary>>} + end; +operate(~"max", Field, Arg, Value) when is_number(Arg) -> + case maps:get(Field, Value, Arg) of + Current when is_number(Current) -> {ok, Value#{Field => max(Current, Arg)}}; + _ -> {error, <<"max on a non-numeric field: ", Field/binary>>} + end; +operate(~"latch", Field, Arg, Value) when is_boolean(Arg) -> + Current = maps:get(Field, Value, false), + {ok, Value#{Field => Current =:= true orelse Arg}}; +operate(OpName, Field, _Arg, _Value) when is_binary(OpName) -> + {error, <>}. + +-spec format_term(term()) -> binary(). +format_term(T) -> + iolist_to_binary(io_lib:format("~0p", [T])). diff --git a/src/asobi_sup.erl b/src/asobi_sup.erl index a05b249..3f94232 100644 --- a/src/asobi_sup.erl +++ b/src/asobi_sup.erl @@ -1,6 +1,8 @@ -module(asobi_sup). -behaviour(supervisor). +-include_lib("kernel/include/logger.hrl"). + -export([start_link/0]). -export([init/1]). -ifdef(TEST). @@ -64,6 +66,7 @@ children(_Engine) -> chat_sup(), tournament_sup(), presence_spec(), + kv_spec(), guest_reaper_spec(), console_session_spec(), lua_game_config_spec(), @@ -241,6 +244,15 @@ presence_spec() -> start => {asobi_presence, start_link, []} }. +%% Owns the `game.kv` table. Nothing else depends on it being up - reads fall +%% through to a miss and writes answer `kv_unavailable` - so it needs no +%% ordering against its siblings. +kv_spec() -> + #{ + id => asobi_kv, + start => {asobi_kv, start_link, []} + }. + oidc_providers_spec() -> #{ id => asobi_oidc_providers, @@ -393,15 +405,78 @@ register_limiters() -> O when is_map(O) -> O; _ -> #{} end, - Opts = maps:merge(DefaultOpts, Overrides), Name = limiter_name(Group), - seki:new_limiter(Name, Opts) + case limiter_opts(maps:merge(DefaultOpts, Overrides)) of + invalid -> + ?LOG_WARNING(#{ + event => rate_limit_override_rejected, + group => Group, + msg => ~"rate_limits override is not a valid limiter; using the default" + }), + start_limiter(Group, Name, limiter_opts(DefaultOpts)); + Opts -> + start_limiter(Group, Name, Opts) + end end, Defaults ), asobi_script_log_limiter:init_table(), ignore. +%% The `invalid` clause is unreachable for the defaults above - they are +%% literals that validate - but limiter_opts/1 cannot say so, and asserting it +%% would turn a future typo in the defaults into a crash at boot. Loud instead: +%% a limiter that quietly never started is a limiter that never limits. +-spec start_limiter(atom(), atom(), seki:limiter_opts() | invalid) -> ok. +start_limiter(Group, _Name, invalid) -> + ?LOG_ERROR(#{ + event => rate_limit_default_invalid, + group => Group, + msg => ~"built-in rate limiter default is not a valid limiter; none started for this group" + }), + ok; +start_limiter(_Group, Name, Opts) -> + _ = seki:new_limiter(Name, Opts), + ok. + +%% The overrides come out of `rate_limits` in sys.config, so the merged map is +%% `#{term() => term()}` however carefully the defaults above are written. +%% Narrowed by validating rather than by asserting: a deployment that writes a +%% string where a number goes should get the default limiter and a log line, +%% not a limiter that silently never limits. +-spec limiter_opts(map()) -> seki:limiter_opts() | invalid. +limiter_opts(Merged) -> + case + { + limiter_algorithm(maps:get(algorithm, Merged, undefined)), + pos_int(maps:get(limit, Merged, undefined)), + pos_int(maps:get(window, Merged, undefined)) + } + of + {A, L, W} when A =/= undefined, L =/= undefined, W =/= undefined -> + maybe_burst(#{algorithm => A, limit => L, window => W}, Merged); + _ -> + invalid + end. + +-spec limiter_algorithm(term()) -> seki:algorithm() | undefined. +limiter_algorithm(token_bucket) -> token_bucket; +limiter_algorithm(sliding_window) -> sliding_window; +limiter_algorithm(gcra) -> gcra; +limiter_algorithm(leaky_bucket) -> leaky_bucket; +limiter_algorithm(_) -> undefined. + +-spec pos_int(term()) -> pos_integer() | undefined. +pos_int(N) when is_integer(N), N > 0 -> N; +pos_int(_) -> undefined. + +-spec maybe_burst(seki:limiter_opts(), map()) -> seki:limiter_opts(). +maybe_burst(Opts, Merged) -> + case pos_int(maps:get(burst, Merged, undefined)) of + undefined -> Opts; + Burst -> Opts#{burst => Burst} + end. + limiter_name(auth) -> asobi_auth_limiter; limiter_name(register) -> asobi_register_limiter; limiter_name(iap) -> asobi_iap_limiter; diff --git a/src/lua/asobi_lua_api.erl b/src/lua/asobi_lua_api.erl index e9d33b3..bd27785 100644 --- a/src/lua/asobi_lua_api.erl +++ b/src/lua/asobi_lua_api.erl @@ -44,6 +44,14 @@ game.storage.get(collection, key) game.storage.set(collection, key, value) game.storage.player_get(player_id, collection, key) game.storage.player_set(player_id, collection, key, value) +game.storage.update(collection, key, ops) -- atomic, operator-based merge +game.storage.player_update(player_id, collection, key, ops) + +-- Node-local KV, scoped to this world/match (in memory, TTL'd, lost on restart) +game.kv.get(collection, key) +game.kv.set(collection, key, value[, ttl_seconds]) +game.kv.merge(collection, key, ops[, ttl_seconds]) -- atomic, returns the merged value +game.kv.delete(collection, key) -- Chat game.chat.send(channel_id, sender_id, content) @@ -205,6 +213,13 @@ core_surface(Ctx) -> {[~"game", ~"storage", ~"set"], write, fun_storage_set()}, {[~"game", ~"storage", ~"player_get"], none, fun_storage_player_get()}, {[~"game", ~"storage", ~"player_set"], write, fun_storage_player_set()}, + {[~"game", ~"storage", ~"update"], write, fun_storage_update()}, + {[~"game", ~"storage", ~"player_update"], write, fun_storage_player_update()}, + %% Node-local KV + {[~"game", ~"kv", ~"get"], none, fun_kv_get(Ctx)}, + {[~"game", ~"kv", ~"set"], write, fun_kv_set(Ctx)}, + {[~"game", ~"kv", ~"merge"], write, fun_kv_merge(Ctx)}, + {[~"game", ~"kv", ~"delete"], write, fun_kv_delete(Ctx)}, %% Chat {[~"game", ~"chat", ~"send"], write, fun_chat_send()}, %% Spatial @@ -917,6 +932,141 @@ fun_storage_player_set() -> end end. +%% widgrensit/asobi#572: an atomic, server-side update, so two zones that both +%% hold an entity across a seam cannot lose each other's writes. Still two +%% round trips to the database, so this suits a per-kill or per-threshold write +%% rather than a per-hit one - `game.kv` is the per-hit answer. +fun_storage_update() -> + fun(Args, St) -> + case decode_args(Args, St) of + [Collection, Key, Ops] when + is_binary(Collection), is_binary(Key), is_map(Ops) + -> + wrap_result(storage_update_ops(Collection, Key, undefined, Ops), St); + _ -> + error_result(~"update requires (collection, key, ops)", St) + end + end. + +fun_storage_player_update() -> + fun(Args, St) -> + case decode_args(Args, St) of + [PlayerId, Collection, Key, Ops] when + is_binary(PlayerId), is_binary(Collection), is_binary(Key), is_map(Ops) + -> + wrap_result(storage_update_ops(Collection, Key, PlayerId, Ops), St); + _ -> + error_result( + ~"player_update requires (player_id, collection, key, ops)", St + ) + end + end. + +%% --- Node-local KV --- +%% +%% Scoped to the calling VM's world or match, not global like game.storage: the +%% state this exists for belongs to one world's entities, and two worlds running +%% the same mode script would otherwise collide on the same collection and key +%% with no way for either to notice. + +fun_kv_get(Ctx) -> + with_kv_scope(Ctx, ~"get requires (collection, key)", fun(Scope, Args, St) -> + case Args of + [Collection, Key] when is_binary(Collection), is_binary(Key) -> + case asobi_kv:get(Scope, Collection, Key) of + {ok, Value} -> {ok, wrap_result({ok, Value}, St)}; + not_found -> {ok, ok_result(nil, St)} + end; + _ -> + bad_args + end + end). + +fun_kv_set(Ctx) -> + with_kv_scope(Ctx, ~"set requires (collection, key, value[, ttl_seconds])", fun( + Scope, Args, St + ) -> + case Args of + [Collection, Key, Value | Rest] when is_binary(Collection), is_binary(Key) -> + %% A table, not a scalar. Everything here is merged field by + %% field, so a scalar has nowhere to go - and coercing one to an + %% empty table would store nothing and say it worked. + case to_storage_value(Value) of + Stored when is_map(Stored) -> + Result = + case ttl_arg(Rest) of + undefined -> asobi_kv:set(Scope, Collection, Key, Stored); + Ttl -> asobi_kv:set(Scope, Collection, Key, Stored, Ttl) + end, + {ok, wrap_result(unit(Result), St)}; + _ -> + {ok, error_result(~"kv.set value must be a table", St)} + end; + _ -> + bad_args + end + end). + +fun_kv_merge(Ctx) -> + with_kv_scope(Ctx, ~"merge requires (collection, key, ops[, ttl_seconds])", fun( + Scope, Args, St + ) -> + case Args of + [Collection, Key, Ops | Rest] when + is_binary(Collection), is_binary(Key), is_map(Ops) + -> + Result = + case ttl_arg(Rest) of + undefined -> asobi_kv:merge(Scope, Collection, Key, Ops); + Ttl -> asobi_kv:merge(Scope, Collection, Key, Ops, Ttl) + end, + {ok, wrap_result(Result, St)}; + _ -> + bad_args + end + end). + +fun_kv_delete(Ctx) -> + with_kv_scope(Ctx, ~"delete requires (collection, key)", fun(Scope, Args, St) -> + case Args of + [Collection, Key] when is_binary(Collection), is_binary(Key) -> + {ok, wrap_result(unit(asobi_kv:delete(Scope, Collection, Key)), St)}; + _ -> + bad_args + end + end). + +%% Every game.kv binding needs the same two things - a scope, and a decode of +%% its arguments - and answers the same way when either is missing. +-spec with_kv_scope(map(), binary(), fun()) -> function(). +with_kv_scope(Ctx, Usage, Body) -> + case kv_scope(Ctx) of + undefined -> + fun(_, St) -> error_result(~"kv not available (no world or match context)", St) end; + Scope -> + fun(Args, St) -> + case Body(Scope, decode_args(Args, St), St) of + {ok, Reply} -> Reply; + bad_args -> error_result(Usage, St) + end + end + end. + +-spec kv_scope(map()) -> binary() | undefined. +kv_scope(Ctx) -> + case maps:get(match_id, Ctx, undefined) of + Id when is_binary(Id) -> Id; + _ -> undefined + end. + +-spec ttl_arg([term()]) -> pos_integer() | undefined. +ttl_arg([N | _]) when is_number(N), N > 0 -> trunc(N); +ttl_arg(_) -> undefined. + +-spec unit(ok | {error, term()}) -> {ok, true} | {error, term()}. +unit(ok) -> {ok, true}; +unit({error, _} = Err) -> Err. + %% --- Chat --- fun_chat_send() -> @@ -1617,6 +1767,105 @@ storage_set(Collection, Key, PlayerId, Value) -> Err end. +%% An atomic operator merge on one storage row (widgrensit/asobi#572). +%% +%% `storage_set/4` is a read-modify-write across two statements with nothing +%% between them, so two writers racing on one key silently lose a write - and +%% for the case this exists for, concurrent writers are the normal case, not an +%% edge one: an entity crossing a seam is legitimately held by two zones at +%% once until it is `rehome_margin` past the boundary, and both tick. +%% +%% The fix is the `version` column the schema already carries. The UPDATE is +%% conditional on the version that was read, so a racing writer's update matches +%% zero rows instead of overwriting; that re-reads and re-applies. The operators +%% are commutative (m:asobi_merge_ops), so re-applying to the newer value is +%% correct rather than merely safe. +%% +%% Bounded rather than unbounded: this runs inside a callback with a millisecond +%% budget, so a hot key answers `storage_conflict` and lets the script decide, +%% instead of spinning the zone. +-define(STORAGE_UPDATE_ATTEMPTS, 5). + +-spec storage_update_ops(binary(), binary(), binary() | undefined, map()) -> + {ok, term()} | {error, term()}. +storage_update_ops(Collection, Key, PlayerId, Ops) -> + storage_update_ops(Collection, Key, PlayerId, Ops, ?STORAGE_UPDATE_ATTEMPTS). + +-spec storage_update_ops(binary(), binary(), binary() | undefined, map(), non_neg_integer()) -> + {ok, term()} | {error, term()}. +storage_update_ops(_Collection, _Key, _PlayerId, _Ops, 0) -> + {error, storage_conflict}; +storage_update_ops(Collection, Key, PlayerId, Ops, Attempts) -> + case storage_find(Collection, Key, PlayerId) of + {ok, Doc} -> + case asobi_merge_ops:apply_ops(storage_value(Doc), Ops) of + {ok, Merged} -> + case storage_cas(Collection, Key, PlayerId, Doc, Merged) of + ok -> + {ok, Merged}; + stale -> + storage_update_ops(Collection, Key, PlayerId, Ops, Attempts - 1); + {error, _} = Err -> + Err + end; + {error, _} = Err -> + Err + end; + {error, not_found} -> + case asobi_merge_ops:apply_ops(#{}, Ops) of + {ok, Merged} -> + case storage_insert(Collection, Key, PlayerId, Merged) of + {ok, _} -> + {ok, Merged}; + %% Lost the race to create the row. The other writer's + %% row is now there to merge into. + {error, _} -> + storage_update_ops(Collection, Key, PlayerId, Ops, Attempts - 1) + end; + {error, _} = Err -> + Err + end; + {error, _} = Err -> + Err + end. + +%% Scoped exactly like storage_find/3 - collection, key and the player +%% namespace - plus the version that was read. Anything else would be a +%% cross-namespace write; see maybe_filter_player/2. +-spec storage_cas(binary(), binary(), binary() | undefined, map(), map()) -> + ok | stale | {error, term()}. +storage_cas(Collection, Key, PlayerId, Doc, Merged) -> + Version = storage_version(Doc), + Q0 = kura_query:from(asobi_storage), + Q1 = kura_query:where(Q0, {collection, Collection}), + Q2 = kura_query:where(Q1, {key, Key}), + Q3 = maybe_filter_player(Q2, PlayerId), + Q4 = kura_query:where(Q3, {version, Version}), + Updates = #{ + value => Merged, + version => Version + 1, + updated_at => calendar:universal_time() + }, + case asobi_repo:update_all(Q4, Updates) of + {ok, 0} -> stale; + {ok, _} -> ok; + {error, _} = Err -> Err + end. + +-spec storage_value(map()) -> map(). +storage_value(Doc) -> + case maps:get(value, Doc, #{}) of + V when is_map(V) -> V; + _ -> #{} + end. + +-spec storage_version(map()) -> integer(). +storage_version(Doc) -> + case maps:get(version, Doc, 1) of + V when is_integer(V) -> V; + _ -> 1 + end. + -spec storage_find(binary(), binary(), binary() | undefined) -> {ok, map()} | {error, term()}. storage_find(Collection, Key, PlayerId) -> diff --git a/src/lua/asobi_lua_surface.erl b/src/lua/asobi_lua_surface.erl index 2c54f8e..d998a72 100644 --- a/src/lua/asobi_lua_surface.erl +++ b/src/lua/asobi_lua_surface.erl @@ -38,6 +38,7 @@ Three things live here, and only here: [~"game", ~"economy"], [~"game", ~"leaderboard"], [~"game", ~"storage"], + [~"game", ~"kv"], [~"game", ~"chat"], [~"game", ~"spatial"], [~"game", ~"zone"], diff --git a/test/asobi_kv_tests.erl b/test/asobi_kv_tests.erl new file mode 100644 index 0000000..151fe53 --- /dev/null +++ b/test/asobi_kv_tests.erl @@ -0,0 +1,102 @@ +-module(asobi_kv_tests). +-include_lib("eunit/include/eunit.hrl"). + +setup() -> + {ok, Pid} = asobi_kv:start_link(), + Pid. + +cleanup(Pid) -> + gen_server:stop(Pid), + application:unset_env(asobi, kv_max_keys), + application:unset_env(asobi, kv_ttl_seconds), + ok. + +kv_test_() -> + {foreach, fun setup/0, fun cleanup/1, [ + fun(_) -> {"set then get", fun set_get/0} end, + fun(_) -> {"get of an absent key", fun get_absent/0} end, + fun(_) -> {"merge creates from an empty map", fun merge_creates/0} end, + fun(_) -> {"merge accumulates across calls", fun merge_accumulates/0} end, + fun(_) -> {"merge returns the merged value", fun merge_returns_value/0} end, + fun(_) -> {"a bad operator is reported, not applied", fun merge_bad_op/0} end, + fun(_) -> {"delete removes the key", fun delete_removes/0} end, + fun(_) -> {"scopes do not collide", fun scopes_isolated/0} end, + fun(_) -> {"an expired entry reads as absent", fun ttl_expires/0} end, + fun(_) -> {"a write refreshes the ttl", fun write_refreshes_ttl/0} end, + fun(_) -> {"a new key past the cap is refused", fun cap_refuses_new_keys/0} end, + fun(_) -> {"an existing key is writable past the cap", fun cap_allows_existing/0} end + ]}. + +set_get() -> + ok = asobi_kv:set(~"w1", ~"ships", ~"s1", #{~"hull" => 100}), + ?assertEqual({ok, #{~"hull" => 100}}, asobi_kv:get(~"w1", ~"ships", ~"s1")). + +get_absent() -> + ?assertEqual(not_found, asobi_kv:get(~"w1", ~"ships", ~"nope")). + +merge_creates() -> + {ok, V} = asobi_kv:merge(~"w1", ~"ships", ~"s1", #{~"kills" => #{~"incr" => 1}}), + ?assertEqual(#{~"kills" => 1}, V). + +merge_accumulates() -> + Ops = #{~"hull" => #{~"incr" => -40}}, + ok = asobi_kv:set(~"w1", ~"ships", ~"s1", #{~"hull" => 100}), + {ok, _} = asobi_kv:merge(~"w1", ~"ships", ~"s1", Ops), + {ok, V} = asobi_kv:merge(~"w1", ~"ships", ~"s1", Ops), + ?assertEqual(#{~"hull" => 20}, V), + ?assertEqual({ok, #{~"hull" => 20}}, asobi_kv:get(~"w1", ~"ships", ~"s1")). + +%% A cast could not answer this, and "did that hit kill it" is the question the +%% caller actually has. +merge_returns_value() -> + ok = asobi_kv:set(~"w1", ~"ships", ~"s1", #{~"hull" => 10}), + {ok, V} = asobi_kv:merge(~"w1", ~"ships", ~"s1", #{ + ~"hull" => #{~"incr" => -10}, ~"dead" => #{~"latch" => true} + }), + ?assertEqual(#{~"hull" => 0, ~"dead" => true}, V). + +merge_bad_op() -> + ok = asobi_kv:set(~"w1", ~"ships", ~"s1", #{~"hull" => 100}), + ?assertMatch({error, _}, asobi_kv:merge(~"w1", ~"ships", ~"s1", #{~"hull" => #{~"nope" => 1}})), + ?assertEqual({ok, #{~"hull" => 100}}, asobi_kv:get(~"w1", ~"ships", ~"s1")). + +delete_removes() -> + ok = asobi_kv:set(~"w1", ~"ships", ~"s1", #{~"hull" => 100}), + ok = asobi_kv:delete(~"w1", ~"ships", ~"s1"), + ?assertEqual(not_found, asobi_kv:get(~"w1", ~"ships", ~"s1")). + +%% Two worlds running the same mode script use the same collection and key. +scopes_isolated() -> + ok = asobi_kv:set(~"w1", ~"ships", ~"s1", #{~"hull" => 100}), + ok = asobi_kv:set(~"w2", ~"ships", ~"s1", #{~"hull" => 50}), + ?assertEqual({ok, #{~"hull" => 100}}, asobi_kv:get(~"w1", ~"ships", ~"s1")), + ?assertEqual({ok, #{~"hull" => 50}}, asobi_kv:get(~"w2", ~"ships", ~"s1")). + +ttl_expires() -> + ok = asobi_kv:set(~"w1", ~"ships", ~"s1", #{~"hull" => 100}, 1), + ?assertMatch({ok, _}, asobi_kv:get(~"w1", ~"ships", ~"s1")), + expire_now(~"w1", ~"ships", ~"s1"), + ?assertEqual(not_found, asobi_kv:get(~"w1", ~"ships", ~"s1")). + +write_refreshes_ttl() -> + ok = asobi_kv:set(~"w1", ~"ships", ~"s1", #{~"hull" => 100}, 1), + expire_now(~"w1", ~"ships", ~"s1"), + %% Merging an expired entry starts from empty, and re-arms the expiry. + {ok, _} = asobi_kv:merge(~"w1", ~"ships", ~"s1", #{~"hull" => #{~"set" => 5}}, 60), + ?assertEqual({ok, #{~"hull" => 5}}, asobi_kv:get(~"w1", ~"ships", ~"s1")). + +cap_refuses_new_keys() -> + application:set_env(asobi, kv_max_keys, 1), + ok = asobi_kv:set(~"w1", ~"ships", ~"s1", #{~"hull" => 100}), + ?assertEqual({error, kv_full}, asobi_kv:set(~"w1", ~"ships", ~"s2", #{~"hull" => 100})). + +cap_allows_existing() -> + application:set_env(asobi, kv_max_keys, 1), + ok = asobi_kv:set(~"w1", ~"ships", ~"s1", #{~"hull" => 100}), + ?assertMatch({ok, _}, asobi_kv:merge(~"w1", ~"ships", ~"s1", #{~"hull" => #{~"incr" => -1}})). + +%% Backdate the row rather than sleeping out a real TTL. +expire_now(Scope, Collection, Key) -> + K = {Scope, Collection, Key}, + [{K, Value, _}] = ets:lookup(asobi_kv, K), + true = ets:insert(asobi_kv, {K, Value, erlang:monotonic_time(millisecond) - 1}). diff --git a/test/asobi_lua_kv_tests.erl b/test/asobi_lua_kv_tests.erl new file mode 100644 index 0000000..6635886 --- /dev/null +++ b/test/asobi_lua_kv_tests.erl @@ -0,0 +1,192 @@ +-module(asobi_lua_kv_tests). +-include_lib("eunit/include/eunit.hrl"). + +%% widgrensit/asobi#572: `game.kv` is the node-local tier for state that has to +%% outlive the zone holding it, at tick rate; `game.storage.update` is the +%% atomic version of the storage read-modify-write for the writes that must +%% survive a restart. + +-spec fixture(string()) -> file:filename_all(). +fixture(Name) -> + case code:lib_dir(asobi) of + {error, _} -> error(asobi_not_loaded); + Dir -> filename:absname(filename:join([Dir, "test", "fixtures", "lua", Name])) + end. + +setup() -> + {ok, Pid} = asobi_kv:start_link(), + Pid. + +cleanup(Pid) -> + gen_server:stop(Pid), + ok. + +install(Ctx) -> + {ok, St0} = asobi_lua_loader:new(fixture("test_match.lua")), + asobi_lua_api:install(Ctx, St0). + +install_world() -> + install(#{vm => world, match_id => ~"test-world", match_pid => self()}). + +eval(Code, St) -> + case luerl:do(Code, St) of + {ok, Results, St1} -> {ok, Results, St1}; + {error, Reason, _} -> {error, Reason} + end. + +lua_kv_test_() -> + {foreach, fun setup/0, fun cleanup/1, [ + fun(_) -> {"kv.set then kv.get", fun kv_set_get/0} end, + fun(_) -> {"kv.get of an absent key answers nil, not an error", fun kv_get_absent/0} end, + fun(_) -> {"kv.merge returns the merged value", fun kv_merge_returns/0} end, + fun(_) -> {"kv.merge accumulates across calls", fun kv_merge_accumulates/0} end, + fun(_) -> {"a bad operator is an error the script can see", fun kv_merge_bad_op/0} end, + fun(_) -> {"kv.delete removes the key", fun kv_delete/0} end, + fun(_) -> {"kv.set refuses a scalar value", fun kv_set_scalar/0} end, + fun(_) -> {"kv rows are scoped to the calling VM's world", fun kv_scoped_to_world/0} end, + fun(_) -> {"kv without a scope errors rather than colliding", fun kv_no_scope/0} end + ]}. + +kv_set_get() -> + St = install_world(), + {ok, [_ | _], St1} = eval("return game.kv.set('ships', 's1', { hull = 100 })", St), + {ok, [Result | _], _} = eval("return game.kv.get('ships', 's1').ok.hull", St1), + ?assertEqual(100, Result). + +kv_get_absent() -> + St = install_world(), + {ok, [Result | _], _} = eval("return game.kv.get('ships', 'nope').ok", St), + ?assertEqual(nil, Result). + +kv_merge_returns() -> + St = install_world(), + Code = + "local r = game.kv.merge('ships', 's1', { hull = { incr = -40 }, dead = { latch = false } })\n" + "return r.ok.hull", + {ok, [Result | _], _} = eval(Code, St), + ?assertEqual(-40, Result). + +kv_merge_accumulates() -> + St = install_world(), + Code = + "game.kv.set('ships', 's1', { hull = 100 })\n" + "game.kv.merge('ships', 's1', { hull = { incr = -40 } })\n" + "local r = game.kv.merge('ships', 's1', { hull = { incr = -40 } })\n" + "return r.ok.hull", + {ok, [Result | _], _} = eval(Code, St), + ?assertEqual(20, Result). + +kv_merge_bad_op() -> + St = install_world(), + {ok, [Error | _], _} = eval("return game.kv.merge('ships', 's1', { hull = 4 }).error", St), + ?assert(is_binary(Error)). + +kv_delete() -> + St = install_world(), + Code = + "game.kv.set('ships', 's1', { hull = 100 })\n" + "game.kv.delete('ships', 's1')\n" + "return game.kv.get('ships', 's1').ok", + {ok, [Result | _], _} = eval(Code, St), + ?assertEqual(nil, Result). + +kv_set_scalar() -> + St = install_world(), + {ok, [Error | _], _} = eval("return game.kv.set('ships', 's1', 42).error", St), + ?assert(is_binary(Error)). + +%% Two worlds running the same mode script write the same collection and key. +kv_scoped_to_world() -> + StA = install(#{vm => world, match_id => ~"world-a", match_pid => self()}), + StB = install(#{vm => world, match_id => ~"world-b", match_pid => self()}), + {ok, _, _} = eval("return game.kv.set('ships', 's1', { hull = 100 })", StA), + {ok, _, _} = eval("return game.kv.set('ships', 's1', { hull = 50 })", StB), + {ok, [A | _], _} = eval("return game.kv.get('ships', 's1').ok.hull", StA), + {ok, [B | _], _} = eval("return game.kv.get('ships', 's1').ok.hull", StB), + ?assertEqual(100, A), + ?assertEqual(50, B). + +kv_no_scope() -> + St = install(#{vm => world, match_pid => self()}), + {ok, [Error | _], _} = eval("return game.kv.get('ships', 's1').error", St), + ?assert(is_binary(Error)). + +%% --- game.storage.update --- + +lua_storage_update_test_() -> + {foreach, fun storage_setup/0, fun storage_cleanup/1, [ + fun(_) -> {"update inserts when the row is absent", fun update_inserts/0} end, + fun(_) -> {"update merges into the row it read", fun update_merges/0} end, + fun(_) -> + {"a racing writer is retried, not overwritten", fun update_retries_on_stale/0} + end, + fun(_) -> {"a hot key gives up rather than spinning", fun update_gives_up/0} end, + fun(_) -> {"a bad operator never reaches the database", fun update_bad_op/0} end + ]}. + +storage_setup() -> + meck:new(asobi_repo, [no_link]), + meck:expect(asobi_repo, all, fun(_) -> {ok, []} end), + meck:expect(asobi_repo, insert, fun(_) -> {ok, #{value => #{}}} end), + meck:expect(asobi_repo, update_all, fun(_, _) -> {ok, 1} end), + ok. + +storage_cleanup(_) -> + meck:unload(asobi_repo), + ok. + +storage_st() -> + install(#{vm => world, match_id => ~"test-world", match_pid => self()}). + +update_inserts() -> + Code = "return game.storage.update('ships', 's1', { kills = { incr = 1 } }).ok.kills", + {ok, [Result | _], _} = eval(Code, storage_st()), + ?assertEqual(1, Result), + ?assert(meck:called(asobi_repo, insert, '_')). + +update_merges() -> + meck:expect(asobi_repo, all, fun(_) -> + {ok, [#{id => ~"row1", value => #{~"hull" => 100}, version => 3}]} + end), + Code = "return game.storage.update('ships', 's1', { hull = { incr = -40 } }).ok.hull", + {ok, [Result | _], _} = eval(Code, storage_st()), + ?assertEqual(60, Result), + %% The UPDATE is conditional on the version that was read - that is what + %% makes this safe for two zones writing the same entity. + ?assert(meck:called(asobi_repo, update_all, '_')), + ?assertNot(meck:called(asobi_repo, insert, '_')). + +update_retries_on_stale() -> + Versions = counters:new(1, []), + meck:expect(asobi_repo, all, fun(_) -> + N = counters:get(Versions, 1), + counters:add(Versions, 1, 1), + {ok, [#{id => ~"row1", value => #{~"hull" => 100 - N * 10}, version => N}]} + end), + meck:expect(asobi_repo, update_all, fun(_, _) -> + case counters:get(Versions, 1) of + 1 -> {ok, 0}; + _ -> {ok, 1} + end + end), + Code = "return game.storage.update('ships', 's1', { hull = { incr = -40 } }).ok.hull", + {ok, [Result | _], _} = eval(Code, storage_st()), + %% Re-read the newer value and re-applied to it, rather than writing over + %% the other writer's damage. + ?assertEqual(50, Result). + +update_gives_up() -> + meck:expect(asobi_repo, all, fun(_) -> + {ok, [#{id => ~"row1", value => #{~"hull" => 100}, version => 3}]} + end), + meck:expect(asobi_repo, update_all, fun(_, _) -> {ok, 0} end), + Code = "return game.storage.update('ships', 's1', { hull = { incr = -40 } }).error", + {ok, [Error | _], _} = eval(Code, storage_st()), + ?assert(is_binary(Error)). + +update_bad_op() -> + Code = "return game.storage.update('ships', 's1', { hull = { nope = 1 } }).error", + {ok, [Error | _], _} = eval(Code, storage_st()), + ?assert(is_binary(Error)), + ?assertNot(meck:called(asobi_repo, insert, '_')), + ?assertNot(meck:called(asobi_repo, update_all, '_')). diff --git a/test/asobi_merge_ops_tests.erl b/test/asobi_merge_ops_tests.erl new file mode 100644 index 0000000..2c51ac2 --- /dev/null +++ b/test/asobi_merge_ops_tests.erl @@ -0,0 +1,82 @@ +-module(asobi_merge_ops_tests). +-include_lib("eunit/include/eunit.hrl"). + +%% widgrensit/asobi#572. The property that matters is not that each operator +%% computes the right number - it is that applying two merges in either order +%% gives the same answer, because two zones holding one entity across a seam +%% tick on two schedulers with no ordering between them. + +set_test() -> + ?assertEqual({ok, #{~"n" => 3}}, asobi_merge_ops:apply_ops(#{}, #{~"n" => #{~"set" => 3}})). + +set_if_absent_test() -> + Ops = #{~"n" => #{~"set_if_absent" => 3}}, + ?assertEqual({ok, #{~"n" => 3}}, asobi_merge_ops:apply_ops(#{}, Ops)), + ?assertEqual({ok, #{~"n" => 1}}, asobi_merge_ops:apply_ops(#{~"n" => 1}, Ops)). + +incr_test() -> + Ops = #{~"hull" => #{~"incr" => -40}}, + {ok, V1} = asobi_merge_ops:apply_ops(#{~"hull" => 100}, Ops), + ?assertEqual(#{~"hull" => 60}, V1), + ?assertEqual({ok, #{~"hull" => -40}}, asobi_merge_ops:apply_ops(#{}, Ops)). + +min_max_test() -> + ?assertEqual( + {ok, #{~"hull" => 40}}, + asobi_merge_ops:apply_ops(#{~"hull" => 100}, #{~"hull" => #{~"min" => 40}}) + ), + ?assertEqual( + {ok, #{~"hull" => 100}}, + asobi_merge_ops:apply_ops(#{~"hull" => 100}, #{~"hull" => #{~"min" => 140}}) + ), + ?assertEqual( + {ok, #{~"shed" => 4}}, + asobi_merge_ops:apply_ops(#{~"shed" => 2}, #{~"shed" => #{~"max" => 4}}) + ), + %% Absent field: the argument is the starting point, not 0. + ?assertEqual( + {ok, #{~"hull" => 40}}, asobi_merge_ops:apply_ops(#{}, #{~"hull" => #{~"min" => 40}}) + ). + +latch_is_sticky_test() -> + On = #{~"dead" => #{~"latch" => true}}, + Off = #{~"dead" => #{~"latch" => false}}, + {ok, V1} = asobi_merge_ops:apply_ops(#{}, Off), + ?assertEqual(#{~"dead" => false}, V1), + {ok, V2} = asobi_merge_ops:apply_ops(V1, On), + ?assertEqual(#{~"dead" => true}, V2), + ?assertEqual({ok, V2}, asobi_merge_ops:apply_ops(V2, Off)). + +%% The whole point: order-independence for the commutative operators. +order_independent_test() -> + A = #{~"hull" => #{~"min" => 60}, ~"shed" => #{~"max" => 2}, ~"kills" => #{~"incr" => 1}}, + B = #{~"hull" => #{~"min" => 40}, ~"shed" => #{~"max" => 4}, ~"kills" => #{~"incr" => 2}}, + Start = #{~"hull" => 100, ~"shed" => 0, ~"kills" => 0}, + {ok, AB0} = asobi_merge_ops:apply_ops(Start, A), + {ok, AB} = asobi_merge_ops:apply_ops(AB0, B), + {ok, BA0} = asobi_merge_ops:apply_ops(Start, B), + {ok, BA} = asobi_merge_ops:apply_ops(BA0, A), + ?assertEqual(AB, BA), + ?assertEqual(#{~"hull" => 40, ~"shed" => 4, ~"kills" => 3}, AB). + +unknown_operator_is_an_error_test() -> + ?assertMatch( + {error, _}, asobi_merge_ops:apply_ops(#{}, #{~"n" => #{~"increment" => 1}}) + ). + +%% A bare value is NOT quietly treated as `set`: that reads exactly like a +%% working merge and loses writes. +bare_value_is_an_error_test() -> + ?assertMatch({error, _}, asobi_merge_ops:apply_ops(#{}, #{~"n" => 3})). + +two_operators_in_one_field_is_an_error_test() -> + Ops = #{~"n" => #{~"min" => 1, ~"max" => 2}}, + ?assertMatch({error, _}, asobi_merge_ops:apply_ops(#{}, Ops)). + +wrong_type_is_an_error_test() -> + ?assertMatch( + {error, _}, + asobi_merge_ops:apply_ops(#{~"n" => ~"text"}, #{~"n" => #{~"incr" => 1}}) + ), + ?assertMatch({error, _}, asobi_merge_ops:apply_ops(#{}, #{~"n" => #{~"incr" => ~"one"}})), + ?assertMatch({error, _}, asobi_merge_ops:apply_ops(#{}, #{~"n" => #{~"latch" => 1}})).