Fix/prefixed lmdb normalize inline

src/core/resolver/hb_cache.erl

@@ -778,6 +778,12 @@ read_resolved_path(Target, ResolvedFullPath, Store, Opts) ->
                     {subpaths, {explicit, Subpaths}}
                 }
             ),
+            Values =
+                maps:from_list([
+                    {hb_util:bin(Subpath), Value}
+                ||
+                    {Subpath, Value} <- Children
+                ]),
             Msg =
                 case hb_opts:get(cache_read_mode, normal, Opts) of
                     raw ->
@@ -786,7 +792,11 @@ read_resolved_path(Target, ResolvedFullPath, Store, Opts) ->
                                 {
                                     Subpath,
                                     {link,
-                                        hb_path:to_binary([ResolvedFullPath, Subpath]),
+                                        child_path(
+                                            ResolvedFullPath,
+                                            Subpath,
+                                            Values
+                                        ),
                                         #{ <<"lazy">> => true, <<"store">> => Store }
                                     }
                                 }
@@ -796,12 +806,6 @@ read_resolved_path(Target, ResolvedFullPath, Store, Opts) ->
                             ]
                         );
                     _ ->
-                        Values =
-                            maps:from_list([
-                                {hb_util:bin(Subpath), Value}
-                            ||
-                                {Subpath, Value} <- Children
-                            ]),
                         prepare_typed_values(
                             Target,
                             ResolvedFullPath,
@@ -829,6 +833,14 @@ read_resolved_path(Target, ResolvedFullPath, Store, Opts) ->
 child_name({Subpath, _Value}) -> Subpath;
 child_name(Subpath) -> Subpath.
 
+%% @doc Use a typed storage target when available; otherwise address the child
+%% beneath its parent. For `+link' children the target contains the message ID.
+child_path(RootPath, Subpath, Values) ->
+    case maps:get(Subpath, Values, none) of
+        {link, Path} -> Path;
+        _ -> hb_path:to_binary([RootPath, Subpath])
+    end.
+
 %% @doc Prepare child values using `ao-types' where present. Immediate values
 %% are returned directly; linked values remain lazy.
 prepare_typed_values(Target, RootPath, Subpaths, Values, Store, Opts) ->
@@ -850,6 +862,8 @@ prepare_typed_values(Target, RootPath, Subpaths, Values, Store, Opts) ->
                     % `RootPath/commitments/Target' path from scratch.
                     CommPath =
                         case maps:get(<<"commitments">>, Values, none) of
+                            {link, CommGroup} ->
+                                hb_path:to_binary([CommGroup, Target]);
                             <<"link:", CommGroup/binary>> when byte_size(CommGroup) > 0 ->
                                 hb_path:to_binary([CommGroup, Target]);
                             _ ->
@@ -894,7 +908,7 @@ prepare_typed_values(Target, RootPath, Subpaths, Values, Store, Opts) ->
                             {subpath, Subpath}
                         }
                     ),
-                    SubkeyPath = hb_path:to_binary([RootPath, Subpath]),
+                    SubkeyPath = child_path(RootPath, Subpath, Values),
                     case hb_link:is_link_key(Subpath) of
                         false ->
                             % The key is a literal value, not a nested composite
@@ -974,7 +988,7 @@ read_ao_types(Path, Subpaths, Values, Store, Opts) ->
     ?event_debug({reading_ao_types, {path, Path}, {subpaths, {explicit, Subpaths}}}),
     case lists:member(<<"ao-types">>, Subpaths) of
         true ->
-            TypesPath = hb_path:to_binary([Path, <<"ao-types">>]),
+            TypesPath = child_path(Path, <<"ao-types">>, Values),
             TypesBin =
                 case Values of
                     #{ <<"ao-types">> := <<"raw:", Direct/binary>> } -> Direct;
@@ -1440,6 +1454,69 @@ match_store_control_test() ->
     ?assertEqual([], match_store(#{ <<"store">> => Store, <<"match-index">> => false })),
     ?assertEqual([], match_store(#{ <<"store">> => Store, <<"match-index">> => [] })).
 
+%% @doc Typed storage targets take precedence, while binary markers and child
+%% names retain the path beneath their parent.
+child_path_test() ->
+    Values = #{
+        <<"child">> => {link, <<"canonical">>},
+        <<"legacy">> => <<"link:legacy">>,
+        <<"nested+link">> => {link, <<"target">>}
+    },
+    ?assertEqual(
+        <<"canonical">>,
+        child_path(<<"root">>, <<"child">>, Values)
+    ),
+    ?assertEqual(
+        <<"target">>,
+        child_path(<<"root">>, <<"nested+link">>, Values)
+    ),
+    ?assertEqual(
+        <<"root/legacy">>,
+        child_path(<<"root">>, <<"legacy">>, Values)
+    ),
+    ?assertEqual(
+        <<"root/missing">>,
+        child_path(<<"root">>, <<"missing">>, Values)
+    ).
+
+%% @doc Canonical storage targets load linked type metadata and nested-message
+%% IDs even when the parent and child names cannot reconstruct those targets.
+canonical_child_targets_test() ->
+    Store = hb_test_utils:test_store(hb_store_volatile, <<"canonical-children">>),
+    Opts = #{ <<"store">> => Store },
+    NestedID = hb_util:encode(crypto:hash(sha256, <<"nested-message">>)),
+    Values = #{
+        <<"ao-types">> => {link, <<"canonical-types">>},
+        <<"count">> => <<"7">>,
+        <<"nested+link">> => {link, <<"canonical-id">>}
+    },
+    ok = hb_store:start(Store),
+    try
+        ok = hb_store:write(Store, #{
+            <<"canonical-types">> => <<"count=\"integer\"">>,
+            <<"canonical-id">> => NestedID,
+            <<NestedID/binary, "/value">> => <<"ok">>
+        }, Opts),
+        Msg = prepare_typed_values(
+            <<"root">>, <<"root">>, maps:keys(Values), Values, Store, Opts
+        ),
+        ?assertEqual(7, maps:get(<<"count">>, Msg)),
+        ?assertEqual(
+            #{ <<"value">> => <<"ok">> },
+            ensure_all_loaded(maps:get(<<"nested">>, Msg), Opts)
+        ),
+        RawPath = child_path(<<"root">>, <<"nested+link">>, Values),
+        ?assertEqual(
+            NestedID,
+            ensure_loaded(
+                {link, RawPath, #{ <<"lazy">> => true }},
+                Opts#{ <<"cache-read-mode">> => raw }
+            )
+        )
+    after
+        hb_store:stop(Store)
+    end.
+
 cache_suite_test_() ->
     hb_store:generate_test_suite([
         {"store unsigned empty message",

src/core/store/hb_store.erl

@@ -33,7 +33,12 @@
 %%%                   request map of the form `#{ <<"list">> => Path }`.
 %%%                   Composite read results may also return children as
 %%%                   `{Key, Value}' pairs when the store can provide a child
-%%%                   value without an additional read.
+%%%                   value without an additional read. `{link, Path}'
+%%%                   identifies a redirect to an absolute, store-native path.
+%%%                   The manager applies `from-key' to the complete target
+%%%                   and restores any stripped prefix before returning it.
+%%%                   Child names remain relative; link targets returned by
+%%%                   the manager are canonical caller-facing paths.
 %%% '''
 %%% Each function takes a `store' message first, containing an arbitrary set
 %%% of its necessary configuration keys, as well as the `store-module' key which
@@ -60,7 +65,7 @@
 %%%     Pre-normalization ->
 %%%     Store invocation ->
 %%%     Post-normalization ->
-%%%     Re-prefixing (only for `resolve`) ->
+%%%     Re-prefixing (`resolve` results and storage-link targets) ->
 %%%     Return.
 %%% '''
 %%%
@@ -606,6 +611,12 @@ from_store(Store, read, {composite, Children}, Opts) ->
 from_store(Store, list, {ok, Children}, Opts) ->
     from_children(Store, Children, Opts);
 from_store(Store, resolve, {ok, Path}, Opts) ->
+    from_absolute_key(Store, Path, Opts);
+from_store(_Store, _Function, Result, _Opts) ->
+    Result.
+
+%% @doc Normalize an absolute path from a store, restoring a stripped prefix.
+from_absolute_key(Store, Path, Opts) ->
     Prefix = maps:get(<<"prefix">>, Store, <<>>),
     maybe
         {ok, Norm} ?= execute_normalizer(<<"from-key">>, Store, Path, Opts),
@@ -615,16 +626,20 @@ from_store(Store, resolve, {ok, Path}, Opts) ->
                 false -> Norm
             end
         }
-    end;
-from_store(_Store, _Function, Result, _Opts) ->
-    Result.
+    end.
 
 %% @doc Normalize the children a store enumerates from it.
 from_children(Store, Children, Opts) ->
     maybe_each(fun(Child) -> from_child(Store, Child, Opts) end, Children).
 
-%% @doc Normalize a child from the store: its key through `from-key', and
-%% the value of a child given as a pair through `from-value'.
+%% @doc Normalize a child from the store: its key through `from-key', a storage
+%% link's target as an absolute key, and an inline value through `from-value'.
+from_child(Store, {Key, {link, Path}}, Opts) ->
+    maybe
+        {ok, NormKey} ?= execute_normalizer(<<"from-key">>, Store, Key, Opts),
+        {ok, NormPath} ?= from_absolute_key(Store, Path, Opts),
+        {ok, {NormKey, {link, NormPath}}}
+    end;
 from_child(Store, {Key, Value}, Opts) ->
     maybe
         {ok, NormKey} ?= execute_normalizer(<<"from-key">>, Store, Key, Opts),
@@ -1338,6 +1353,17 @@ benchmark_message(nested, N, TestDataSize) ->
 
 %%% Normalization Pipeline Tests
 
+%% @doc A storage-link target regains its prefix, while its child name does not.
+storage_link_normalization_test() ->
+    ?assertEqual(
+        {ok, {<<"child">>, {link, <<"mnt/target">>}}},
+        from_child(
+            #{ <<"prefix">> => <<"mnt/">> },
+            {<<"child">>, {link, <<"target">>}},
+            #{}
+        )
+    ).
+
 %% @doc Test that a store with a `prefix' only admits keys bearing it --
 %% writes and reads alike -- stripping the prefix ahead of invocation and
 %% falling through to later stores otherwise, and that a path resolved
@@ -1385,6 +1411,104 @@ prefix_pipeline_test() ->
     ?assertEqual({ok, <<"2">>}, read([Plain], <<"outer">>, #{})),
     ?event(testing, {unprefixed_skip_and_strip_off_passed}).
 
+%% @doc A mounted LMDB child's value is the same through direct and parent
+%% reads, even when a later store contains its unprefixed target.
+prefix_pipeline_lmdb_link_test() ->
+    Mounted =
+        (hb_test_utils:test_store(hb_store_lmdb, <<"pipeline-link">>))#{
+            <<"prefix">> => <<"mnt/">>
+        },
+    Plain = hb_test_utils:test_store(hb_store_volatile, <<"pipeline-link-plain">>),
+    Opts = #{
+        <<"store">> => [Mounted, Plain],
+        <<"cache-control">> => [<<"no-cache">>, <<"no-store">>]
+    },
+    ok = start([Mounted, Plain]),
+    try
+        ok = write(#{ <<"mnt/payload">> => <<"mounted">> }, Opts),
+        ok = link(#{ <<"mnt/msg/body">> => <<"mnt/payload">> }, Opts),
+        ok = write(#{ <<"payload">> => <<"unrelated">> }, Opts),
+        ?assertEqual(
+            {composite, [{<<"body">>, {link, <<"payload">>}}]},
+            hb_store_lmdb:read(Mounted, #{ <<"read">> => <<"msg">> }, Opts)
+        ),
+        lists:foreach(
+            fun(Stores) ->
+                ReadOpts = Opts#{ <<"store">> => Stores },
+                ?assertEqual(
+                    {ok, <<"mounted">>},
+                    hb_cache:read(<<"mnt/msg/body">>, ReadOpts)
+                ),
+                ?assertEqual(
+                    {composite,
+                        [{<<"body">>, {link, <<"mnt/payload">>}}]},
+                    read(Stores, <<"mnt/msg">>, ReadOpts)
+                ),
+                {ok, Msg} = hb_cache:read(<<"mnt/msg">>, ReadOpts),
+                ?assertMatch(
+                    {link, <<"mnt/payload">>, _},
+                    maps:get(<<"body">>, Msg)
+                ),
+                ?assertEqual(
+                    {ok, <<"mounted">>},
+                    hb_ao:resolve(Msg, <<"body">>, ReadOpts)
+                )
+            end,
+            [[Mounted, Plain], [Mounted]]
+        )
+    after
+        stop([Mounted, Plain])
+    end.
+
+%% @doc LMDB inline link targets are normalized as complete absolute paths;
+%% they are not rebuilt from independently normalized parent and child keys.
+normalize_pipeline_lmdb_link_test() ->
+    Store =
+        (hb_test_utils:test_store(
+            hb_store_lmdb,
+            <<"pipeline-normalized-link">>
+        ))#{
+            <<"prefix">> => <<"b64/">>,
+            <<"to-key">> => <<"~base64url@1.0/decode/body">>,
+            <<"from-key">> => <<"~base64url@1.0/encode/body">>
+        },
+    Opts = #{
+        <<"store">> => Store,
+        <<"cache-control">> => [<<"no-cache">>, <<"no-store">>]
+    },
+    Canonical =
+        fun(Path) -> <<"b64/", (hb_util:encode(Path))/binary>> end,
+    PayloadPath = Canonical(<<"payload">>),
+    MessagePath = Canonical(<<"msg">>),
+    BodyPath = Canonical(<<"msg/body">>),
+    BodyKey = hb_util:encode(<<"body">>),
+    ok = start(Store),
+    try
+        ok = write(#{ PayloadPath => <<"mounted">> }, Opts),
+        ok = link(#{ BodyPath => PayloadPath }, Opts),
+        {ok, Msg} = hb_cache:read(MessagePath, Opts),
+        ?assertMatch(
+            {link, PayloadPath, _},
+            maps:get(BodyKey, Msg)
+        ),
+        ?assertEqual(
+            {ok, <<"mounted">>},
+            hb_ao:resolve(Msg, BodyKey, Opts)
+        ),
+        RawOpts = Opts#{ <<"cache-read-mode">> => raw },
+        {ok, RawMsg} = hb_cache:read(MessagePath, RawOpts),
+        ?assertMatch(
+            {link, PayloadPath, _},
+            maps:get(BodyKey, RawMsg)
+        ),
+        ?assertEqual(
+            {ok, <<"mounted">>},
+            hb_ao:resolve(RawMsg, BodyKey, RawOpts)
+        )
+    after
+        stop(Store)
+    end.
+
 %% @doc Test that lifecycle operations bypass path preprocessing for a store
 %% carrying a prefix.
 prefix_pipeline_stop_test() ->

src/core/store/hb_store_lmdb.erl

@@ -525,7 +525,7 @@ immediate_children(Prefix, Rows) ->
             case Key of
                 <<Prefix:PrefixSize/binary, Child/binary>> when Child =/= <<>> ->
                     case binary:match(Child, <<"/">>) of
-                        nomatch -> {true, {Child, Value}};
+                        nomatch -> {true, {Child, child_value(Value)}};
                         _ -> false
                     end;
                 _ ->
@@ -535,6 +535,13 @@ immediate_children(Prefix, Rows) ->
         Rows
     ).
 
+%% @doc Expose LMDB redirects as typed storage links in composite reads.
+child_value(Value) ->
+    case is_link(Value) of
+        {true, Path} -> {link, Path};
+        false -> Value
+    end.
+
 child_prefix(<<>>) -> <<>>;
 child_prefix(<<"/">>) -> <<>>;
 child_prefix(Path) ->
@@ -1244,3 +1251,11 @@ read_prefix_composite_test() ->
     ),
     ?assertEqual({ok, [<<"a">>, <<"b">>]}, test_list(StoreOpts, <<"root">>)),
     test_stop(StoreOpts).
+
+%% @doc Redirects are typed, while escaped literal marker values stay binary.
+child_value_test() ->
+    ?assertEqual(
+        {link, <<"target">>},
+        child_value(<<"link:target">>)
+    ),
+    ?assertEqual(<<"raw:link:literal">>, child_value(<<"raw:link:literal">>)).