search

daemons/node/common/common_pubsub.go

@@ -19,7 +19,8 @@ type Event struct {
 	Type string `json:"type"`
 	From string `json:"from"` // peerID
 
-	User string
+	User   string
+	Groups []string
 
 	DataType  int64  `json:"datatype"`
 	Timestamp int64  `json:"ts"`

daemons/node/common/search_tracker.go

@@ -0,0 +1,94 @@
+package common
+
+import (
+	"context"
+	"oc-discovery/conf"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/google/uuid"
+)
+
+// SearchIdleTimeout returns the configured search idle timeout (default 5s).
+func SearchIdleTimeout() time.Duration {
+	if t := conf.GetConfig().SearchTimeout; t > 0 {
+		return time.Duration(t) * time.Second
+	}
+	return 5 * time.Second
+}
+
+// searchEntry holds the lifecycle state for one active search.
+type searchEntry struct {
+	cancel      context.CancelFunc
+	timer       *time.Timer
+	idleTimeout time.Duration
+}
+
+// SearchTracker tracks one active search per user (peer or resource).
+// Each search is keyed by a composite "user:searchID" so that a replaced
+// search's late-arriving results can be told apart from the current one.
+//
+// Typical usage:
+//
+//	ctx, cancel := context.WithCancel(parent)
+//	key := tracker.Register(userKey, cancel, idleTimeout)
+//	defer tracker.Cancel(key)
+//	// ... on each result: tracker.ResetIdle(key) + tracker.IsActive(key)
+type SearchTracker struct {
+	mu      sync.Mutex
+	entries map[string]*searchEntry
+}
+
+func NewSearchTracker() *SearchTracker {
+	return &SearchTracker{entries: map[string]*searchEntry{}}
+}
+
+// Register starts a new search for baseUser, cancelling any previous one.
+// Returns the composite key "baseUser:searchID" to be used as the search identifier.
+func (t *SearchTracker) Register(baseUser string, cancel context.CancelFunc, idleTimeout time.Duration) string {
+	compositeKey := baseUser + ":" + uuid.New().String()
+	t.mu.Lock()
+	t.cancelByPrefix(baseUser)
+	e := &searchEntry{cancel: cancel, idleTimeout: idleTimeout}
+	e.timer = time.AfterFunc(idleTimeout, func() { t.Cancel(compositeKey) })
+	t.entries[compositeKey] = e
+	t.mu.Unlock()
+	return compositeKey
+}
+
+// Cancel cancels the search(es) matching user (bare user key or composite key).
+func (t *SearchTracker) Cancel(user string) {
+	t.mu.Lock()
+	t.cancelByPrefix(user)
+	t.mu.Unlock()
+}
+
+// ResetIdle resets the idle timer for compositeKey after a response arrives.
+func (t *SearchTracker) ResetIdle(compositeKey string) {
+	t.mu.Lock()
+	if e, ok := t.entries[compositeKey]; ok {
+		e.timer.Reset(e.idleTimeout)
+	}
+	t.mu.Unlock()
+}
+
+// IsActive returns true if compositeKey is still the current active search.
+func (t *SearchTracker) IsActive(compositeKey string) bool {
+	t.mu.Lock()
+	_, ok := t.entries[compositeKey]
+	t.mu.Unlock()
+	return ok
+}
+
+// cancelByPrefix cancels all entries whose key equals user or starts with "user:".
+// Must be called with t.mu held.
+func (t *SearchTracker) cancelByPrefix(user string) {
+	for k, e := range t.entries {
+		if k == user || strings.HasPrefix(k, user+":") {
+			e.timer.Stop()
+			e.cancel()
+			delete(t.entries, k)
+		}
+	}
+}

daemons/node/nats.go

@@ -105,7 +105,7 @@ func ListenNATS(n *Node) {
 				dtt := tools.DataType(propalgation.DataType)
 				dt = &dtt
 			}
-			fmt.Println("PROPALGATION ACT", propalgation.Action, propalgation.Action == tools.PB_CREATE, err)
+			fmt.Println("PROPALGATION ACT", propalgation.DataType, propalgation.Action, propalgation.Action == tools.PB_CREATE, err)
 			if err == nil {
 				switch propalgation.Action {
 				case tools.PB_ADMIRALTY_CONFIG, tools.PB_MINIO_CONFIG:
@@ -117,16 +117,15 @@ func ListenNATS(n *Node) {
 					if err := json.Unmarshal(resp.Payload, &m); err == nil {
 						peers, _ := n.GetPeerRecord(context.Background(), m.PeerID)
 						for _, p := range peers {
-							n.StreamService.PublishCommon(&resp.Datatype, resp.User,
+							n.StreamService.PublishCommon(&resp.Datatype, resp.User, resp.Groups,
 								p.PeerID, proto, resp.Payload)
 						}
 					}
 				case tools.PB_CREATE, tools.PB_UPDATE, tools.PB_DELETE:
-					fmt.Println(propalgation.Action, dt, resp.User, propalgation.Payload)
 					fmt.Println(n.StreamService.ToPartnerPublishEvent(
 						context.Background(),
 						propalgation.Action,
-						dt, resp.User,
+						dt, resp.User, resp.Groups,
 						propalgation.Payload,
 					))
 				case tools.PB_CONSIDERS:
@@ -137,7 +136,7 @@ func ListenNATS(n *Node) {
 							for _, p := range m.PeerIDs {
 								peers, _ := n.GetPeerRecord(context.Background(), p)
 								for _, pp := range peers {
-									n.StreamService.PublishCommon(&resp.Datatype, resp.User,
+									n.StreamService.PublishCommon(&resp.Datatype, resp.User, resp.Groups,
 										pp.PeerID, stream.ProtocolConsidersResource, resp.Payload)
 								}
 							}
@@ -150,7 +149,7 @@ func ListenNATS(n *Node) {
 						if err := json.Unmarshal(propalgation.Payload, &m); err == nil && m.OriginID != "" {
 							peers, _ := n.GetPeerRecord(context.Background(), m.OriginID)
 							for _, p := range peers {
-								n.StreamService.PublishCommon(nil, resp.User,
+								n.StreamService.PublishCommon(nil, resp.User, resp.Groups,
 									p.PeerID, stream.ProtocolConsidersResource, propalgation.Payload)
 							}
 						}
@@ -166,11 +165,11 @@ func ListenNATS(n *Node) {
 							n.StreamService.Mu.Lock()
 							if n.StreamService.Streams[stream.ProtocolSendPlanner] != nil {
 								for pid := range n.StreamService.Streams[stream.ProtocolSendPlanner] { // send Planner can be long lived - it's a conn
-									n.StreamService.PublishCommon(nil, resp.User, pid.String(), stream.ProtocolSendPlanner, b)
+									n.StreamService.PublishCommon(nil, resp.User, resp.Groups, pid.String(), stream.ProtocolSendPlanner, b)
 								}
 							}
 						} else {
-							n.StreamService.PublishCommon(nil, resp.User, fmt.Sprintf("%v", m["peer_id"]), stream.ProtocolSendPlanner, b)
+							n.StreamService.PublishCommon(nil, resp.User, resp.Groups, fmt.Sprintf("%v", m["peer_id"]), stream.ProtocolSendPlanner, b)
 						}
 						n.StreamService.Mu.Unlock()
 					}
@@ -186,6 +185,12 @@ func ListenNATS(n *Node) {
 						}
 						n.StreamService.Mu.Unlock()
 					}
+				case tools.PB_CLOSE_SEARCH:
+					if propalgation.DataType == int(tools.PEER) {
+						n.peerSearches.Cancel(resp.User)
+					} else {
+						n.StreamService.ResourceSearches.Cancel(resp.User)
+					}
 				case tools.PB_SEARCH:
 					if propalgation.DataType == int(tools.PEER) {
 						m := map[string]interface{}{}
@@ -211,7 +216,7 @@ func ListenNATS(n *Node) {
 								context.Background(),
 								dt,
 								fmt.Sprintf("%v", m["type"]),
-								resp.User,
+								resp.User, resp.Groups,
 								fmt.Sprintf("%v", m["search"]),
 							)
 						}

daemons/node/node.go

@@ -27,12 +27,6 @@ import (
 	"github.com/libp2p/go-libp2p/p2p/security/noise"
 )
 
-// activeSearch tracks an in-flight distributed peer search for one user.
-type activeSearch struct {
-	queryID string
-	cancel  context.CancelFunc
-}
-
 type Node struct {
 	*common.LongLivedStreamRecordedService[interface{}] // change type of stream
 	PS                                                  *pubsubs.PubSub
@@ -43,9 +37,8 @@ type Node struct {
 	isIndexer                                           bool
 	peerRecord                                          *indexer.PeerRecord
 
-	// activeSearches: one streaming search per user; new search cancels previous.
-	activeSearchesMu sync.Mutex
-	activeSearches   map[string]*activeSearch
+	// peerSearches: one active peer search per user; new search cancels previous.
+	peerSearches *common.SearchTracker
 
 	Mu sync.RWMutex
 }
@@ -85,7 +78,7 @@ func InitNode(isNode bool, isIndexer bool) (*Node, error) {
 		PeerID:                         h.ID(),
 		isIndexer:                      isIndexer,
 		LongLivedStreamRecordedService: common.NewStreamRecordedService[interface{}](h, 1000),
-		activeSearches:                 map[string]*activeSearch{},
+		peerSearches:                   common.NewSearchTracker(),
 	}
 	// Register the bandwidth probe handler so any peer measuring this node's
 	// throughput can open a dedicated probe stream and read the echo.
@@ -204,32 +197,21 @@ func (d *Node) publishPeerRecord(
 }
 
 // SearchPeerRecord starts a distributed peer search via ProtocolSearchPeer.
-// userKey identifies the requesting user — a new call cancels any previous
-// search for the same user. Results are pushed to onResult as they arrive.
-// The function returns when the search stream closes (idle timeout or indexer unreachable).
+// A new call for the same userKey cancels any previous search.
+// Results are pushed to onResult as they arrive; the function returns when
+// the stream closes (idle timeout, explicit cancel, or indexer unreachable).
 func (d *Node) SearchPeerRecord(userKey, needle string, onResult func(common.SearchHit)) {
+	fmt.Println("SearchPeerRecord", needle)
 	logger := oclib.GetLogger()
 
+	idleTimeout := common.SearchIdleTimeout()
 	ctx, cancel := context.WithCancel(context.Background())
+	// Register cancels any previous search for userKey and starts the idle timer.
+	// The composite key doubles as QueryID so the indexer echoes it back.
+	searchKey := d.peerSearches.Register(userKey, cancel, idleTimeout)
+	defer d.peerSearches.Cancel(searchKey)
 
-	d.activeSearchesMu.Lock()
-	if prev, ok := d.activeSearches[userKey]; ok {
-		prev.cancel()
-	}
-	queryID := uuid.New().String()
-	d.activeSearches[userKey] = &activeSearch{queryID: queryID, cancel: cancel}
-	d.activeSearchesMu.Unlock()
-
-	defer func() {
-		cancel()
-		d.activeSearchesMu.Lock()
-		if cur, ok := d.activeSearches[userKey]; ok && cur.queryID == queryID {
-			delete(d.activeSearches, userKey)
-		}
-		d.activeSearchesMu.Unlock()
-	}()
-
-	req := common.SearchPeerRequest{QueryID: queryID}
+	req := common.SearchPeerRequest{QueryID: searchKey}
 	if pid, err := pp.Decode(needle); err == nil {
 		req.PeerID = pid.String()
 	} else if _, err := uuid.Parse(needle); err == nil {
@@ -238,7 +220,6 @@ func (d *Node) SearchPeerRecord(userKey, needle string, onResult func(common.Sea
 		req.Name = needle
 	}
 
-	// Try indexers in pool order until one accepts the stream.
 	for _, ad := range common.Indexers.GetAddrs() {
 		if ad.Info == nil {
 			continue
@@ -253,15 +234,22 @@ func (d *Node) SearchPeerRecord(userKey, needle string, onResult func(common.Sea
 			s.Reset()
 			continue
 		}
+		// Interrupt the blocking Decode as soon as the context is cancelled
+		// (idle timer, explicit PB_CLOSE_SEARCH, or replacement search).
+		go func() {
+			<-ctx.Done()
+			s.SetReadDeadline(time.Now())
+		}()
 		dec := json.NewDecoder(s)
 		for {
 			var result common.SearchPeerResult
 			if err := dec.Decode(&result); err != nil {
 				break
 			}
-			if result.QueryID != queryID {
-				continue // stale response from a previous query
+			if result.QueryID != searchKey || !d.peerSearches.IsActive(searchKey) {
+				break
 			}
+			d.peerSearches.ResetIdle(searchKey)
 			for _, hit := range result.Records {
 				onResult(hit)
 			}

daemons/node/pubsub/publish.go

@@ -4,9 +4,11 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"oc-discovery/conf"
 	"oc-discovery/daemons/node/common"
 	"oc-discovery/daemons/node/stream"
 	"oc-discovery/models"
+	"time"
 
 	"cloud.o-forge.io/core/oc-lib/dbs"
 	"cloud.o-forge.io/core/oc-lib/models/peer"
@@ -14,16 +16,16 @@ import (
 )
 
 func (ps *PubSubService) SearchPublishEvent(
-	ctx context.Context, dt *tools.DataType, typ string, user string, search string) error {
+	ctx context.Context, dt *tools.DataType, typ string, user string, groups []string, search string) error {
 	b, err := json.Marshal(map[string]string{"search": search})
 	if err != nil {
 		return err
 	}
 	switch typ {
 	case "known": // define Search Strategy
-		return ps.StreamService.PublishesCommon(dt, user, nil, b, stream.ProtocolSearchResource) //if partners focus only them*/
+		return ps.StreamService.PublishesCommon(dt, user, groups, nil, b, stream.ProtocolSearchResource) //if partners focus only them*/
 	case "partner": // define Search Strategy
-		return ps.StreamService.PublishesCommon(dt, user, &dbs.Filters{ // filter by like name, short_description, description, owner, url if no filters are provided
+		return ps.StreamService.PublishesCommon(dt, user, groups, &dbs.Filters{ // filter by like name, short_description, description, owner, url if no filters are provided
 			And: map[string][]dbs.Filter{
 				"relation": {{Operator: dbs.EQUAL.String(), Value: peer.PARTNER}},
 			},
@@ -33,7 +35,18 @@ func (ps *PubSubService) SearchPublishEvent(
 		if err != nil {
 			return err
 		}
-		return ps.publishEvent(ctx, dt, tools.PB_SEARCH, common.TopicPubSubSearch, user, b)
+		idleTimeout := func() time.Duration {
+			if t := conf.GetConfig().SearchTimeout; t > 0 {
+				return time.Duration(t) * time.Second
+			}
+			return 5 * time.Second
+		}()
+		searchCtx, cancel := context.WithCancel(ctx)
+		// Register cancels any previous search for this user and starts the idle timer.
+		// The returned composite key is used as User in the GossipSub event so that
+		// remote peers echo it back unchanged, allowing IsActive to validate results.
+		searchKey := ps.StreamService.ResourceSearches.Register(user, cancel, idleTimeout)
+		return ps.publishEvent(searchCtx, dt, tools.PB_SEARCH, common.TopicPubSubSearch, searchKey, b)
 	default:
 		return errors.New("no type of research found")
 	}

daemons/node/stream/handler.go

@@ -80,7 +80,7 @@ func (abs *StreamService) verifyResponse(event *common.Event) error { //
 		}
 	}
 	if b, err := json.Marshal(verify); err == nil {
-		abs.PublishCommon(nil, "", event.From, ProtocolVerifyResource, b)
+		abs.PublishCommon(nil, event.User, event.Groups, event.From, ProtocolVerifyResource, b)
 	}
 	return nil
 }
@@ -89,7 +89,7 @@ func (abs *StreamService) sendPlanner(event *common.Event) error { //
 	if len(event.Payload) == 0 {
 		if plan, err := planner.GenerateShallow(&tools.APIRequest{Admin: true}); err == nil {
 			if b, err := json.Marshal(plan); err == nil {
-				abs.PublishCommon(nil, event.User, event.From, ProtocolSendPlanner, b)
+				abs.PublishCommon(nil, event.User, event.Groups, event.From, ProtocolSendPlanner, b)
 			} else {
 				return err
 			}
@@ -114,10 +114,15 @@ func (abs *StreamService) sendPlanner(event *common.Event) error { //
 }
 
 func (abs *StreamService) retrieveResponse(event *common.Event) error { //
+	if !abs.ResourceSearches.IsActive(event.User) {
+		return nil // search already closed or timed out
+	}
 	res, err := resources.ToResource(int(event.DataType), event.Payload)
 	if err != nil || res == nil {
 		return nil
 	}
+	// A response arrived — reset the idle timeout.
+	abs.ResourceSearches.ResetIdle(event.User)
 	b, err := json.Marshal(res.Serialize(res))
 	go tools.NewNATSCaller().SetNATSPub(tools.SEARCH_EVENT, tools.NATSResponse{
 		FromApp:  "oc-discovery",
@@ -214,11 +219,11 @@ func (abs *StreamService) SendResponse(p *peer.Peer, event *common.Event, search
 		for _, dt := range dts {
 			access := oclib.NewRequestAdmin(oclib.LibDataEnum(dt), nil)
 			peerID := p.GetID()
-			searched := access.Search(abs.FilterPeer(self.GetID(), search), "", false)
+			searched := access.Search(abs.FilterPeer(self.GetID(), event.Groups, search), "", false)
 			fmt.Println("SEND SEARCH_EVENT", self.GetID(), dt, len(searched.Data), peerID)
 			for _, ss := range searched.Data {
 				if j, err := json.Marshal(ss); err == nil {
-					_, err := abs.PublishCommon(&dt, event.User, p.PeerID, ProtocolSearchResource, j)
+					_, err := abs.PublishCommon(&dt, event.User, event.Groups, p.PeerID, ProtocolSearchResource, j)
 					fmt.Println("Publish ERR", err)
 				}
 			}

daemons/node/stream/publish.go

@@ -15,7 +15,7 @@ import (
 	"github.com/libp2p/go-libp2p/core/protocol"
 )
 
-func (ps *StreamService) PublishesCommon(dt *tools.DataType, user string, filter *dbs.Filters, resource []byte, protos ...protocol.ID) error {
+func (ps *StreamService) PublishesCommon(dt *tools.DataType, user string, groups []string, filter *dbs.Filters, resource []byte, protos ...protocol.ID) error {
 	access := oclib.NewRequestAdmin(oclib.LibDataEnum(oclib.PEER), nil)
 	var p oclib.LibDataShallow
 	if filter == nil {
@@ -25,7 +25,7 @@ func (ps *StreamService) PublishesCommon(dt *tools.DataType, user string, filter
 	}
 	for _, pes := range p.Data {
 		for _, proto := range protos {
-			if _, err := ps.PublishCommon(dt, user, pes.(*peer.Peer).PeerID, proto, resource); err != nil {
+			if _, err := ps.PublishCommon(dt, user, groups, pes.(*peer.Peer).PeerID, proto, resource); err != nil {
 				return err
 			}
 		}
@@ -33,7 +33,7 @@ func (ps *StreamService) PublishesCommon(dt *tools.DataType, user string, filter
 	return nil
 }
 
-func (ps *StreamService) PublishCommon(dt *tools.DataType, user string, toPeerID string, proto protocol.ID, resource []byte) (*common.Stream, error) {
+func (ps *StreamService) PublishCommon(dt *tools.DataType, user string, groups []string, toPeerID string, proto protocol.ID, resource []byte) (*common.Stream, error) {
 	fmt.Println("PublishCommon")
 	if toPeerID == ps.Key.String() {
 		fmt.Println("Can't send to ourself !")
@@ -64,7 +64,7 @@ func (ps *StreamService) PublishCommon(dt *tools.DataType, user string, toPeerID
 }
 
 func (ps *StreamService) ToPartnerPublishEvent(
-	ctx context.Context, action tools.PubSubAction, dt *tools.DataType, user string, payload []byte) error {
+	ctx context.Context, action tools.PubSubAction, dt *tools.DataType, user string, groups []string, payload []byte) error {
 	if *dt == tools.PEER {
 		var p peer.Peer
 		if err := json.Unmarshal(payload, &p); err != nil {
@@ -82,7 +82,7 @@ func (ps *StreamService) ToPartnerPublishEvent(
 			pe.Relation = p.Relation
 			pe.Verify = false
 			if b2, err := json.Marshal(pe); err == nil {
-				if _, err := ps.PublishCommon(dt, user, p.PeerID, ProtocolUpdateResource, b2); err != nil {
+				if _, err := ps.PublishCommon(dt, user, groups, p.PeerID, ProtocolUpdateResource, b2); err != nil {
 					return err
 				}
 
@@ -102,7 +102,7 @@ func (ps *StreamService) ToPartnerPublishEvent(
 	case tools.PB_UPDATE:
 		proto = ProtocolUpdateResource
 	}
-	ps.PublishesCommon(dt, user, &dbs.Filters{ // filter by like name, short_description, description, owner, url if no filters are provided
+	ps.PublishesCommon(dt, user, groups, &dbs.Filters{ // filter by like name, short_description, description, owner, url if no filters are provided
 		And: map[string][]dbs.Filter{
 			"relation": {{Operator: dbs.EQUAL.String(), Value: peer.PARTNER}},
 		},

daemons/node/stream/service.go

@@ -51,22 +51,24 @@ var protocolsPartners = map[protocol.ID]*common.ProtocolInfo{
 }
 
 type StreamService struct {
-	Key          pp.ID
-	Host         host.Host
-	Node         common.DiscoveryPeer
-	Streams       common.ProtocolStream
-	maxNodesConn int
-	Mu           sync.RWMutex
+	Key             pp.ID
+	Host            host.Host
+	Node            common.DiscoveryPeer
+	Streams         common.ProtocolStream
+	maxNodesConn    int
+	Mu              sync.RWMutex
+	ResourceSearches *common.SearchTracker
 }
 
 func InitStream(ctx context.Context, h host.Host, key pp.ID, maxNode int, node common.DiscoveryPeer) (*StreamService, error) {
 	logger := oclib.GetLogger()
 	service := &StreamService{
-		Key:          key,
-		Node:         node,
-		Host:         h,
-		Streams:       common.ProtocolStream{},
-		maxNodesConn: maxNode,
+		Key:              key,
+		Node:             node,
+		Host:             h,
+		Streams:          common.ProtocolStream{},
+		maxNodesConn:     maxNode,
+		ResourceSearches: common.NewSearchTracker(),
 	}
 	for proto := range protocols {
 		service.Host.SetStreamHandler(proto, service.HandleResponse)
@@ -101,7 +103,6 @@ func (s *StreamService) HandleResponse(stream network.Stream) {
 		stream.Protocol(), protocols[stream.Protocol()])
 }
 
-
 func (s *StreamService) connectToPartners() error {
 	logger := oclib.GetLogger()
 	// Register handlers for partner resource protocols (create/update/delete).
@@ -123,7 +124,6 @@ func (s *StreamService) connectToPartners() error {
 	return nil
 }
 
-
 func (s *StreamService) searchPeer(search string) ([]*peer.Peer, error) {
 	ps := []*peer.Peer{}
 	if conf.GetConfig().PeerIDS != "" {
@@ -219,11 +219,12 @@ func (ps *StreamService) readLoop(s *common.Stream, id pp.ID, proto protocol.ID,
 	}
 }
 
-func (abs *StreamService) FilterPeer(peerID string, search string) *dbs.Filters {
+func (abs *StreamService) FilterPeer(peerID string, groups []string, search string) *dbs.Filters {
 	p, err := oclib.GetMySelf()
 	if err != nil {
 		return nil
 	}
+	groups = append(groups, "*")
 	filter := map[string][]dbs.Filter{
 		"abstractinstanciatedresource.abstractresource.abstractobject.creator_id": {{Operator: dbs.EQUAL.String(), Value: p.GetID()}}, // is my resource...
 		"": {{Operator: dbs.OR.String(), Value: &dbs.Filters{
@@ -233,7 +234,7 @@ func (abs *StreamService) FilterPeer(peerID string, search string) *dbs.Filters
 					And: map[string][]dbs.Filter{
 						"resourceinstance.partnerships": {{Operator: dbs.ELEMMATCH.String(), Value: &dbs.Filters{
 							And: map[string][]dbs.Filter{
-								"resourcepartnership.peer_groups." + peerID: {{Operator: dbs.EXISTS.String(), Value: true}},
+								"resourcepartnership.peer_groups." + peerID: {{Operator: dbs.IN.String(), Value: groups}},
 							},
 						}}},
 					},

go.mod

@@ -3,7 +3,7 @@ module oc-discovery
 go 1.25.0
 
 require (
-	cloud.o-forge.io/core/oc-lib v0.0.0-20260311072518-933b7147e908
+	cloud.o-forge.io/core/oc-lib v0.0.0-20260312141150-a335c905b3a2
 	github.com/ipfs/go-cid v0.6.0
 	github.com/libp2p/go-libp2p v0.47.0
 	github.com/libp2p/go-libp2p-record v0.3.1

go.sum

@@ -2,6 +2,10 @@ cloud.o-forge.io/core/oc-lib v0.0.0-20260304145747-e03a0d3dd0aa h1:1wCpI4dwN1pj6
 cloud.o-forge.io/core/oc-lib v0.0.0-20260304145747-e03a0d3dd0aa/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA=
 cloud.o-forge.io/core/oc-lib v0.0.0-20260311072518-933b7147e908 h1:1jz3xI/u2FzCG8phY7ShqADrmCj0mlrdjbdNUosSwgs=
 cloud.o-forge.io/core/oc-lib v0.0.0-20260311072518-933b7147e908/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260312073634-2c9c42dd516a h1:oCkb9l/Cvn0x6iicxIydrjfCNU+UHhKuklFgfzDa174=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260312073634-2c9c42dd516a/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260312141150-a335c905b3a2 h1:DuB6SDThFVJVQ0iI0pZnBqtCE0uW+SNI7R7ndKixu2k=
+cloud.o-forge.io/core/oc-lib v0.0.0-20260312141150-a335c905b3a2/go.mod h1:+ENuvBfZdESSvecoqGY/wSvRlT3vinEolxKgwbOhUpA=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/Masterminds/semver/v3 v3.4.0 h1:Zog+i5UMtVoCU8oKka5P7i9q9HgrJeGzI9SA1Xbatp0=
 github.com/Masterminds/semver/v3 v3.4.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM=