Repository: skupperproject/skupper
Pull request #1543 (2024-07-15 20:41:16)

Vanflow foundations
The code in pkg/flow has become tricky to adapat as it is coupled to the network-console api and large parts of it are gated based on a collector "mode". This is a large departure that moves to a new set of vanflow primitives that I have mostly copied from an independent implementation I had done (https://github.com/c-kruse/vanflow) save for a few adaptations made from lessons learned. My goals here were to make it sustainable to develop multiple vanflow components independently and to allow each piece to be better tested.

@grs
I've got a few comments commit-by-commit.

4e1b91c0275e69a6eaad126ca6224dc2205dfd5b: This is the new set of primitives and is the most polished bit of this draft PR. Maybe noteworthy/contentious, adds two new dependencies that I thought were sensible when this started as an independent implementation - the azure vcabbage fork, and the exponential backoff module I was most familiar with. Also includes a new class of unit/integration tests that were super useful for me: tests ran against a real skupper router (most of which fall back on a mock router.) Not sure if those have a future or not as they are likely to be flakey/resource  dependent.

e263854212548fb09b37d479977484e49b7e8147: Replaces the controller in pkg/flow with a new one in pkg/kube/flow. This was the first and most straight-forward vanflow component to rewrite (most all of the work is done by a `eventsource.Manager`.) I am sure I've made a bit of a mess of the controller and its interface just from not really understanding the existing controller constructs (it was hard to handle deletes since the work queue + cache key off namespace/name and I'm storing records by object uid: by the time I understood that I had done my own controller loop.)

e9519f67af51d8440cc5f1c78955d2978bf8d152: A new "collector-lite" implementation that I have called pkg/kube/flow.StatusSync. The interface is still rough and my plans for it not totally realized (e.g. no need to use vanflow to get records from the controller running in the same process). Was feeling less confident on this one so it is running alongside the old implementation presently: would be interesting we could find an easy way to continuously diff their output. That aside - I hope it is easy enough to reason. My goal was that any of us could pretty confidently manage to adapt context from vanflow records into the network status info.


**EDIT**: Now that I think of it: perhaps both components I put in pkg/kube/flow should get rewritten as a single controller.

c-kruse: > The link records only seem to have the name in them at present.

@grs Expected! Was going to sort that out as a followup. I suspect the right thing to do is going to end up being adding the new Link counterpart, RouterAccess, to network status.

Patch:
diff --git a/cmd/config-sync/collector.go b/cmd/config-sync/collector.go
index dbd026c7..d21303f9 100644
--- a/cmd/config-sync/collector.go
+++ b/cmd/config-sync/collector.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"log/slog"
 	"os"
 	"time"
 
@@ -14,8 +15,11 @@ import (
 	"github.com/skupperproject/skupper/pkg/kube"
 	kubeflow "github.com/skupperproject/skupper/pkg/kube/flow"
 	"github.com/skupperproject/skupper/pkg/qdr"
-	"github.com/skupperproject/skupper/pkg/version"
+	"github.com/skupperproject/skupper/pkg/vanflow"
+	"github.com/skupperproject/skupper/pkg/vanflow/session"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	corev1informer "k8s.io/client-go/informers/core/v1"
+	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/leaderelection"
 	"k8s.io/client-go/tools/leaderelection/resourcelock"
 )
@@ -47,16 +51,23 @@ func siteCollector(stopCh <-chan struct{}, cli *internalclient.KubeClient) {
 	}
 	owner := kube.GetDeploymentOwnerReference(current)
 
+	const shadowNetworkStatusConfigMapName = "skupper-network-status-shadow"
 	existing, err := kube.NewConfigMap(types.NetworkStatusConfigMapName, &siteData, nil, nil, &owner, cli.Namespace, cli.Kube)
 	if err != nil && existing == nil {
 		log.Fatal("Failed to create site status config map ", err.Error())
 	}
+	existing, err = kube.NewConfigMap(shadowNetworkStatusConfigMapName, &siteData, nil, nil, &owner, cli.Namespace, cli.Kube)
+	if err != nil && existing == nil {
+		log.Fatal("Failed to create site status config map ", err.Error())
+	}
 
 	err = updateLockOwner(types.SiteLeaderLockName, cli.Namespace, &owner, cli)
 	if err != nil {
 		log.Println("Update lock error", err.Error())
 	}
 
+	factory := session.NewContainerFactory("amqp://localhost:5672", session.ContainerConfig{ContainerID: "kube-flow-collector"})
+	statusSync := kubeflow.NewStatusSync(factory, nil, cli.Kube.CoreV1().ConfigMaps(cli.Namespace), shadowNetworkStatusConfigMapName)
 	fc = flow.NewFlowCollector(flow.FlowCollectorSpec{
 		Mode:                flow.RecordStatus,
 		Namespace:           cli.Namespace,
@@ -68,7 +79,9 @@ func siteCollector(stopCh <-chan struct{}, cli *internalclient.KubeClient) {
 
 	go primeBeacons(fc, cli)
 	log.Println("COLLECTOR: Starting flow collector")
+	go statusSync.Run(context.Background())
 	fc.Start(stopCh)
+
 }
 
 // primeBeacons attempts to guess the router and service-controller vanflow IDs
@@ -97,15 +110,34 @@ func startFlowController(stopCh <-chan struct{}, cli *internalclient.KubeClient)
 	if err != nil {
 		log.Fatal("Failed to get transport deployment", err.Error())
 	}
-	creationTime := uint64(deployment.ObjectMeta.CreationTimestamp.UnixNano()) / uint64(time.Microsecond)
-	flowController := flow.NewFlowController(siteId, version.Version, creationTime, qdr.NewConnectionFactory("amqp://localhost:5672", nil), nil /*TODO: enable policy checks?*/)
 
-	controller := kube.NewController("flow-controller", cli)
-	kubeflow.WatchPods(controller, cli.Namespace, flowController.UpdateProcess)
+	informer := corev1informer.NewPodInformer(cli.Kube, cli.Namespace, time.Minute*5, cache.Indexers{})
+	platform := "kubernetes"
+	fc := kubeflow.NewController(kubeflow.ControllerConfig{
+		Factory:  session.NewContainerFactory("amqp://localhost:5672", session.ContainerConfig{ContainerID: "kube-flow-controller"}),
+		Informer: informer,
+		Site: vanflow.SiteRecord{
+			BaseRecord: vanflow.NewBase(siteId, deployment.ObjectMeta.CreationTimestamp.Time),
+			Name:       &deployment.Name,
+			Namespace:  &cli.Namespace,
+			Platform:   &platform,
+		},
+	})
+	go informer.Run(stopCh)
 	//TODO: should watching nodes be optional or should we attempt to determine if we have permissions first?
 	//kubeflow.WatchNodes(controller, cli.Namespace, flowController.UpdateHost)
-	controller.Start(stopCh)
-	flowController.Start(stopCh)
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		<-stopCh
+		cancel()
+	}()
+	go func() {
+		defer cancel()
+		fc.Run(ctx)
+		if ctx.Err() == nil {
+			slog.Error("kube flow controller unexpectedly quit")
+		}
+	}()
 	return nil
 }
 
diff --git a/go.mod b/go.mod
index a7eb4381..58c1f730 100644
--- a/go.mod
+++ b/go.mod
@@ -3,7 +3,9 @@ module github.com/skupperproject/skupper
 go 1.21.0
 
 require (
+	github.com/Azure/go-amqp v1.0.5
 	github.com/briandowns/spinner v1.23.0
+	github.com/cenkalti/backoff/v4 v4.3.0
 	github.com/davecgh/go-spew v1.1.1
 	github.com/fsnotify/fsnotify v1.6.0
 	github.com/go-openapi/errors v0.20.3
@@ -12,15 +14,14 @@ require (
 	github.com/go-openapi/swag v0.21.1
 	github.com/go-openapi/validate v0.22.0
 	github.com/golang/glog v1.1.0
-	github.com/google/go-cmp v0.5.9
+	github.com/google/go-cmp v0.6.0
 	github.com/google/uuid v1.3.0
-	github.com/gorilla/handlers v1.5.1
+	github.com/gorilla/handlers v1.5.2
 	github.com/gorilla/mux v1.8.0
 	github.com/interconnectedcloud/go-amqp v0.12.6-0.20200506124159-f51e540008b5
 	github.com/openshift/api v0.0.0-20210428205234-a8389931bee7
 	github.com/openshift/client-go v0.0.0-20210112165513-ebc401615f47
 	github.com/prometheus/client_golang v1.14.0
-	github.com/rogpeppe/go-internal v1.3.0
 	github.com/skupperproject/skupper-example-tcp-echo v0.0.0-20210727195922-db4a7dc7b35d
 	github.com/skupperproject/skupper-libpod/v4 v4.0.3-0
 	github.com/spf13/cobra v1.8.0
@@ -29,6 +30,7 @@ require (
 	go.mongodb.org/mongo-driver v1.10.0
 	golang.org/x/crypto v0.14.0
 	golang.org/x/net v0.17.0
+	golang.org/x/sys v0.14.0
 	golang.org/x/text v0.13.0
 	google.golang.org/grpc v1.56.3
 	google.golang.org/protobuf v1.30.0
@@ -37,6 +39,7 @@ require (
 	k8s.io/api v0.20.0
 	k8s.io/apimachinery v0.20.0
 	k8s.io/client-go v0.20.0
+	sigs.k8s.io/yaml v1.2.0
 )
 
 replace github.com/gogo/protobuf => github.com/gogo/protobuf v1.3.2
@@ -64,7 +67,6 @@ require (
 	github.com/fatih/color v1.7.0 // indirect
 	github.com/felixge/httpsnoop v1.0.3 // indirect
 	github.com/form3tech-oss/jwt-go v3.2.2+incompatible // indirect
-	github.com/fortytw2/leaktest v1.3.0 // indirect
 	github.com/go-logr/logr v0.2.0 // indirect
 	github.com/go-openapi/analysis v0.21.2 // indirect
 	github.com/go-openapi/jsonpointer v0.19.5 // indirect
@@ -105,7 +107,6 @@ require (
 	github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d // indirect
 	golang.org/x/oauth2 v0.7.0 // indirect
 	golang.org/x/sync v0.1.0 // indirect
-	golang.org/x/sys v0.14.0 // indirect
 	golang.org/x/term v0.14.0 // indirect
 	golang.org/x/time v0.0.0-20200630173020-3af7569d3a1e // indirect
 	google.golang.org/appengine v1.6.7 // indirect
@@ -116,5 +117,4 @@ require (
 	k8s.io/kube-openapi v0.0.0-20201113171705-d219536bb9fd // indirect
 	k8s.io/utils v0.0.0-20201110183641-67b214c5f920 // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.0.2 // indirect
-	sigs.k8s.io/yaml v1.2.0 // indirect
 )
diff --git a/go.sum b/go.sum
index d43e179a..05206553 100644
--- a/go.sum
+++ b/go.sum
@@ -37,6 +37,8 @@ cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 github.com/Azure/azure-sdk-for-go v46.0.0+incompatible h1:4qlEOCDcDQZTGczYGzbGYCdJfVpZLIs8AEo5+MoXBPw=
 github.com/Azure/azure-sdk-for-go v46.0.0+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc=
+github.com/Azure/go-amqp v1.0.5 h1:po5+ljlcNSU8xtapHTe8gIc8yHxCzC03E8afH2g1ftU=
+github.com/Azure/go-amqp v1.0.5/go.mod h1:vZAogwdrkbyK3Mla8m/CxSc/aKdnTZ4IbPxl51Y5WZE=
 github.com/Azure/go-autorest v14.2.0+incompatible h1:V5VMDjClD3GiElqLWO7mz2MxNAK/vTfRHdAubSIPRgs=
 github.com/Azure/go-autorest v14.2.0+incompatible/go.mod h1:r+4oMnoxhatjLLJ6zxSWATqVooLgysK6ZNox3g/xq24=
 github.com/Azure/go-autorest/autorest v0.11.1 h1:eVvIXUKiTgv++6YnWb42DUA1YL7qDugnKP0HljexdnQ=
@@ -83,6 +85,8 @@ github.com/bmizerany/perks v0.0.0-20141205001514-d9a9656a3a4b/go.mod h1:ac9efd0D
 github.com/briandowns/spinner v1.23.0 h1:alDF2guRWqa/FOZZYWjlMIx2L6H0wyewPxo/CH4Pt2A=
 github.com/briandowns/spinner v1.23.0/go.mod h1:rPG4gmXeN3wQV/TsAY4w8lPdIM6RX3yqeBQJSrbXjuE=
 github.com/c2h5oh/datasize v0.0.0-20171227191756-4eba002a5eae/go.mod h1:S/7n9copUssQ56c7aAgHqftWO4LTf4xY6CGWt8Bc+3M=
+github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
+github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
@@ -119,7 +123,6 @@ github.com/evanphx/json-patch v4.9.0+incompatible h1:kLcOMZeuLAJvL2BPWLMIj5oaZQo
 github.com/evanphx/json-patch v4.9.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk=
 github.com/fatih/color v1.7.0 h1:DkWD4oS2D8LGGgTQ6IvwJJXSL5Vp2ffcQg58nFV38Ys=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
-github.com/felixge/httpsnoop v1.0.1/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
 github.com/felixge/httpsnoop v1.0.3 h1:s/nj+GCswXYzN5v2DpNMuMQYe+0DDwt5WVCU6CWBdXk=
 github.com/felixge/httpsnoop v1.0.3/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
 github.com/form3tech-oss/jwt-go v3.2.2+incompatible h1:TcekIExNqud5crz4xD2pavyTgWiPvpYe4Xau31I0PRk=
@@ -270,8 +273,8 @@ github.com/google/go-cmp v0.5.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
-github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
-github.com/google/go-cmp v0.5.9/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
+github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
+github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/gofuzz v1.1.0 h1:Hsa8mG0dQ46ij8Sl2AYJDUv1oA9/d6Vk+3LG99Oe02g=
 github.com/google/gofuzz v1.1.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
@@ -293,8 +296,8 @@ github.com/googleapis/gax-go/v2 v2.0.4/go.mod h1:0Wqv26UfaUD9n4G6kQubkQ+KchISgw+
 github.com/googleapis/gax-go/v2 v2.0.5/go.mod h1:DWXyrwAJ9X0FpwwEdw+IPEYBICEFu5mhpdKc/us6bOk=
 github.com/googleapis/gnostic v0.4.1 h1:DLJCy1n/vrD4HPjOvYcT8aYQXpPIzoRZONaYwyycI+I=
 github.com/googleapis/gnostic v0.4.1/go.mod h1:LRhVm6pbyptWbWbuZ38d1eyptfvIytN3ir6b65WBswg=
-github.com/gorilla/handlers v1.5.1 h1:9lRY6j8DEeeBT10CvO9hGW0gmky0BprnvDI5vfhUHH4=
-github.com/gorilla/handlers v1.5.1/go.mod h1:t8XrUpc4KVXb7HGyJ4/cEnwQiaxrX/hz1Zv/4g96P1Q=
+github.com/gorilla/handlers v1.5.2 h1:cLTUSsNkgcwhgRqvCNmdbRWG0A3N4F+M2nWKdScwyEE=
+github.com/gorilla/handlers v1.5.2/go.mod h1:dX+xVpaxdSw+q0Qek8SSsl3dfMk3jNddUkMzo0GtH0w=
 github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=
 github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
 github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:FecbI9+v66THATjSRHfNgh1IVFe/9kFxbXtjV0ctIMA=
@@ -432,7 +435,6 @@ github.com/prometheus/procfs v0.8.0 h1:ODq8ZFEaYeCaZOJlZZdJA2AbQR98dSHSM1KW/You5
 github.com/prometheus/procfs v0.8.0/go.mod h1:z7EfXMXOkbkqb9IINtpCn86r/to3BnA0uaxHdg830/4=
 github.com/rogpeppe/go-internal v1.1.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.2.2/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
-github.com/rogpeppe/go-internal v1.3.0 h1:RR9dF3JtopPvtkroDZuVD7qquD0bnHlKSqaQhgwt8yk=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/russross/blackfriday/v2 v2.1.0 h1:JIOH55/0cWyOuilr9/qlrm0BSXldqnqwMsf35Ld67mk=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
@@ -465,8 +467,9 @@ github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81P
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
+github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
 github.com/tidwall/pretty v1.0.0 h1:HsD+QiTn7sK6flMKIvNmpqz1qrpP3Ps6jOKIKMooyg4=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
 github.com/tsenart/go-tsz v0.0.0-20180814232043-cdeb9e1e981e/go.mod h1:SWZznP1z5Ki7hDT2ioqiFKEse8K9tU2OUvaRI0NeGQo=
diff --git a/pkg/kube/flow/controller.go b/pkg/kube/flow/controller.go
new file mode 100644
index 00000000..2b84b3d0
--- /dev/null
+++ b/pkg/kube/flow/controller.go
@@ -0,0 +1,215 @@
+package flow
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"github.com/skupperproject/skupper/pkg/vanflow"
+	"github.com/skupperproject/skupper/pkg/vanflow/eventsource"
+	"github.com/skupperproject/skupper/pkg/vanflow/session"
+	"github.com/skupperproject/skupper/pkg/vanflow/store"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
+)
+
+type ControllerConfig struct {
+	Factory  session.ContainerFactory
+	Site     vanflow.SiteRecord
+	Informer cache.SharedIndexInformer
+}
+
+func NewController(cfg ControllerConfig) *Controller {
+	source := store.SourceRef{
+		ID:      cfg.Site.ID,
+		Version: "1",
+	}
+	staticRecords := store.NewSyncMapStore(store.SyncMapStoreConfig{})
+	staticRecords.Add(cfg.Site, source)
+
+	pods := store.NewSyncMapStore(store.SyncMapStoreConfig{})
+
+	container := cfg.Factory.Create()
+	manager := eventsource.NewManager(container, eventsource.ManagerConfig{
+		Source: eventsource.Info{
+			ID:      cfg.Site.ID,
+			Version: 1,
+			Type:    "CONTROLLER",
+			Address: fmt.Sprintf("mc/sfe.%s", cfg.Site.ID),
+			Direct:  fmt.Sprintf("sfe.%s", cfg.Site.ID),
+		},
+		Stores: []store.Interface{staticRecords, pods},
+
+		UseAlternateHeartbeatAddress: true,
+		FlushDelay:                   time.Millisecond * 100,
+		FlushBatchSize:               20,
+		UpdateBufferTime:             time.Millisecond * 1000,
+		UpdateBatchSize:              10,
+	})
+
+	ctrlr := &Controller{
+		podStore:  pods,
+		podCache:  cfg.Informer.GetStore(),
+		container: container,
+		source:    source,
+		manager:   manager,
+		queue:     workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "kube-flow-controller"),
+		logger: slog.New(slog.Default().Handler()).With(
+			slog.String("component", "kube.flow.controller"),
+		),
+	}
+
+	cfg.Informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj any) {
+			evnt, err := keyFunc(obj)
+			if err != nil {
+				ctrlr.logger.Error("add event error", slog.Any("error", err))
+			}
+			evnt.Type = eventTypeProcess
+			ctrlr.queue.Add(evnt)
+		},
+		UpdateFunc: func(prev, obj any) {
+			evnt, err := keyFunc(obj)
+			if err != nil {
+				ctrlr.logger.Error("update event error", slog.Any("error", err))
+			}
+			evnt.Type = eventTypeProcess
+			ctrlr.queue.Add(evnt)
+		},
+		DeleteFunc: func(obj any) {
+			if final, ok := obj.(cache.DeletedFinalStateUnknown); ok {
+				obj = final.Obj
+			}
+			evnt, err := keyFunc(obj)
+			if err != nil {
+				ctrlr.logger.Error("update event error", slog.Any("error", err))
+			}
+			evnt.Type = eventTypeProcess
+			ctrlr.queue.Add(evnt)
+		},
+	})
+	return ctrlr
+}
+
+type Controller struct {
+	container session.Container
+	podStore  store.Interface
+	podCache  cache.Store
+	source    store.SourceRef
+	manager   *eventsource.Manager
+	queue     workqueue.RateLimitingInterface
+	logger    *slog.Logger
+}
+
+func (c *Controller) handlePodEvent(e workEvent) error {
+	obj, exists, err := c.podCache.GetByKey(e.CacheKey)
+	if err != nil {
+		return err
+	}
+	var record vanflow.ProcessRecord
+	record.ID = e.StoreKey
+	if exists {
+		record = asProcessRecord(obj.(*v1.Pod))
+	}
+	c.updateProcess(!exists, record)
+	return nil
+}
+
+func (c *Controller) Run(ctx context.Context) {
+	c.container.Start(ctx)
+	mgmtCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	c.container.OnSessionError(func(err error) {
+		_, retryable := err.(session.RetryableError)
+		if !retryable {
+			cancel()
+		}
+		c.logger.Error("amqp session error", slog.Any("error", err), slog.Bool("retryable", retryable))
+	})
+	go c.manager.Run(mgmtCtx)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			item, done := c.queue.Get()
+			if done {
+				c.logger.Info("workqueue shutdown")
+				return
+			}
+			evnt := item.(workEvent)
+			switch evnt.Type {
+			case eventTypeProcess:
+				if err := c.handlePodEvent(evnt); err != nil {
+					c.queue.AddRateLimited(evnt)
+				}
+			}
+			c.queue.Forget(item)
+			c.queue.Done(item)
+		}
+	}
+}
+
+func (c *Controller) updateProcess(deleted bool, process vanflow.ProcessRecord) {
+	c.logger.Debug("update process", slog.Bool("deleted", deleted), slog.String("process", process.ID))
+	process.Parent = &c.source.ID
+	if deleted {
+		entry, ok := c.podStore.Delete(process.ID)
+		if !ok {
+			c.logger.Debug("ignoring notification for unknown pod", slog.String("process", process.ID))
+			return
+		}
+		terminalRecord := entry.Record.(vanflow.ProcessRecord)
+		terminalRecord.EndTime = &vanflow.Time{Time: time.Now()}
+		c.manager.PublishUpdate(eventsource.RecordUpdate{
+			Prev: entry.Record,
+			Curr: terminalRecord,
+		})
+		return
+	}
+	var prev vanflow.Record
+	if curr, exists := c.podStore.Get(process.ID); exists {
+		c.podStore.Update(process)
+		prev = curr.Record
+	} else {
+		c.podStore.Add(process, c.source)
+	}
+	c.manager.PublishUpdate(eventsource.RecordUpdate{
+		Prev: prev,
+		Curr: process,
+	})
+}
+
+type workEvent struct {
+	CacheKey string
+	StoreKey string
+	Type     eventType
+}
+
+type eventType string
+
+const (
+	eventTypeProcess eventType = "ProcessRecord"
+)
+
+func keyFunc(obj any) (workEvent, error) {
+	var (
+		out workEvent
+		err error
+	)
+	out.CacheKey, err = cache.MetaNamespaceKeyFunc(obj)
+	if err != nil {
+		return out, fmt.Errorf("failed to get object key: %v", err)
+	}
+	ma, ok := obj.(metav1.ObjectMetaAccessor)
+	if !ok {
+		return out, fmt.Errorf("failed to get object metadata for type %T", obj)
+	}
+	meta := ma.GetObjectMeta()
+	out.StoreKey = string(meta.GetUID())
+
+	return out, nil
+}
diff --git a/pkg/kube/flow/host.go b/pkg/kube/flow/host.go
deleted file mode 100644
index d1425839..00000000
--- a/pkg/kube/flow/host.go
+++ /dev/null
@@ -1,50 +0,0 @@
-package flow
-
-import (
-	"strings"
-	"time"
-
-	corev1 "k8s.io/api/core/v1"
-
-	"github.com/skupperproject/skupper/pkg/flow"
-	"github.com/skupperproject/skupper/pkg/kube"
-)
-
-type HostUpdateHandler func(deleted bool, name string, process *flow.HostRecord) error
-
-func WatchNodes(controller *kube.Controller, namespace string, handler HostUpdateHandler) {
-	controller.WatchNodes(func(key string, node *corev1.Node) error {
-		if node == nil {
-			handler(true, key, nil)
-		} else {
-			handler(false, key, AsHostRecord("-"+namespace, node))
-		}
-		return nil
-	})
-}
-
-func AsHostRecord(qualifier string, node *corev1.Node) *flow.HostRecord {
-	host := &flow.HostRecord{}
-	// Note: skupper running in multiple ns in same cluster the hosts are the same
-	host.Identity = string(node.ObjectMeta.UID) + qualifier
-	host.StartTime = uint64(node.ObjectMeta.CreationTimestamp.UnixNano()) / uint64(time.Microsecond)
-	host.Name = &node.ObjectMeta.Name
-	host.Arch = &node.Status.NodeInfo.Architecture
-
-	provider := strings.Split(node.Spec.ProviderID, "://")
-	if len(provider) > 0 {
-		host.Provider = &provider[0]
-	}
-	if region, ok := node.ObjectMeta.Labels["topology.kubernetes.io/region"]; ok {
-		host.Location = &region
-		host.Region = &region
-	}
-	if zone, ok := node.ObjectMeta.Labels["topology.kubernetes.io/zone"]; ok {
-		host.Zone = &zone
-	}
-	host.ContainerRuntime = &node.Status.NodeInfo.ContainerRuntimeVersion
-	host.KernelVersion = &node.Status.NodeInfo.KernelVersion
-	host.KubeProxyVersion = &node.Status.NodeInfo.KubeProxyVersion
-	host.KubeletVersion = &node.Status.NodeInfo.KubeletVersion
-	return host
-}
diff --git a/pkg/kube/flow/indexing.go b/pkg/kube/flow/indexing.go
new file mode 100644
index 00000000..1116bead
--- /dev/null
+++ b/pkg/kube/flow/indexing.go
@@ -0,0 +1,63 @@
+package flow
+
+import (
+	"fmt"
+
+	"github.com/skupperproject/skupper/pkg/vanflow"
+	"github.com/skupperproject/skupper/pkg/vanflow/store"
+)
+
+const (
+	byTypeParent = "ByTypeAndParent"
+	byAddress    = "ByAddress"
+	byParentHost = "ByParentHost"
+)
+
+func indexByParentHost(e store.Entry) []string {
+	if proc, ok := e.Record.(vanflow.ProcessRecord); ok {
+		if proc.Parent != nil && proc.SourceHost != nil {
+			return []string{fmt.Sprintf("%s/%s", *proc.Parent, *proc.SourceHost)}
+		}
+	}
+	return nil
+}
+func indexByTypeParent(e store.Entry) []string {
+	optionalSingle := func(prefix string, s *string) []string {
+		if s != nil {
+			return []string{fmt.Sprintf("%s/%s", prefix, *s)}
+		}
+		return nil
+	}
+	switch record := e.Record.(type) {
+	case vanflow.RouterRecord:
+		return optionalSingle(record.GetTypeMeta().String(), record.Parent)
+	case vanflow.LinkRecord:
+		return optionalSingle(record.GetTypeMeta().String(), record.Parent)
+	case vanflow.RouterAccessRecord:
+		return optionalSingle(record.GetTypeMeta().String(), record.Parent)
+	case vanflow.ConnectorRecord:
+		return optionalSingle(record.GetTypeMeta().String(), record.Parent)
+	case vanflow.ListenerRecord:
+		return optionalSingle(record.GetTypeMeta().String(), record.Parent)
+	case vanflow.ProcessRecord:
+		return optionalSingle(record.GetTypeMeta().String(), record.Parent)
+	default:
+		return nil
+	}
+}
+func indexByAddress(e store.Entry) []string {
+	optionalSingle := func(s *string) []string {
+		if s != nil {
+			return []string{*s}
+		}
+		return nil
+	}
+	switch record := e.Record.(type) {
+	case vanflow.ConnectorRecord:
+		return optionalSingle(record.Address)
+	case vanflow.ListenerRecord:
+		return optionalSingle(record.Address)
+	default:
+		return nil
+	}
+}
diff --git a/pkg/kube/flow/process.go b/pkg/kube/flow/process.go
index 798bc327..5f05f053 100644
--- a/pkg/kube/flow/process.go
+++ b/pkg/kube/flow/process.go
@@ -2,54 +2,43 @@ package flow
 
 import (
 	"strings"
-	"time"
 
 	corev1 "k8s.io/api/core/v1"
 
-	"github.com/skupperproject/skupper/pkg/flow"
-	"github.com/skupperproject/skupper/pkg/kube"
+	"github.com/skupperproject/skupper/pkg/vanflow"
 )
 
-type ProcessUpdateHandler func(deleted bool, name string, process *flow.ProcessRecord) error
-
-func WatchPods(controller *kube.Controller, namespace string, handler ProcessUpdateHandler) {
-	controller.WatchAllPods(namespace, func(key string, pod *corev1.Pod) error {
-		if pod == nil {
-			handler(true, key, nil)
-		} else {
-			handler(false, key, AsProcessRecord(pod))
-		}
-		return nil
-	})
-}
+var (
+	modeExternal = "external"
+	modeInternal = "internal"
+)
 
-func AsProcessRecord(pod *corev1.Pod) *flow.ProcessRecord {
-	process := &flow.ProcessRecord{}
-	process.Identity = string(pod.ObjectMeta.UID)
-	process.StartTime = uint64(pod.ObjectMeta.CreationTimestamp.UnixNano()) / uint64(time.Microsecond)
-	process.Name = &pod.ObjectMeta.Name
+func asProcessRecord(pod *corev1.Pod) vanflow.ProcessRecord {
+	process := vanflow.ProcessRecord{
+		BaseRecord: vanflow.NewBase(string(pod.ObjectMeta.UID), pod.ObjectMeta.CreationTimestamp.Time),
+		Name:       &pod.ObjectMeta.Name,
+		Mode:       &modeExternal,
+	}
 	if pod.Status.PodIP != "" {
 		process.SourceHost = &pod.Status.PodIP
 	}
-	process.Image = &pod.Spec.Containers[0].Image
-	process.ImageName = process.Image
-	process.HostName = &pod.Spec.NodeName
-	process.ProcessRole = &flow.External
+	process.ImageName = &pod.Spec.Containers[0].Image
+	process.Hostname = &pod.Spec.NodeName
 	if labelName, ok := pod.ObjectMeta.Labels["app.kubernetes.io/part-of"]; ok {
-		process.GroupName = &labelName
+		process.Group = &labelName
 		if labelName == "skupper" {
-			process.ProcessRole = &flow.Internal
+			process.Mode = &modeInternal
 		}
 	} else if labelComponent, ok := pod.ObjectMeta.Labels["app.kubernetes.io/name"]; ok {
-		process.GroupName = &labelComponent
+		process.Group = &labelComponent
 	} else if partOf, ok := pod.ObjectMeta.Labels["app.kubernetes.io/component"]; ok {
-		process.GroupName = &partOf
+		process.Group = &partOf
 	} else {
 		// generate process group from image name
 		parts := strings.Split(*process.ImageName, "/")
 		part := parts[len(parts)-1]
 		pg := strings.Split(part, ":")
-		process.GroupName = &pg[0]
+		process.Group = &pg[0]
 	}
 	return process
 }
diff --git a/pkg/kube/flow/status.go b/pkg/kube/flow/status.go
new file mode 100644
index 00000000..36aecb15
--- /dev/null
+++ b/pkg/kube/flow/status.go
@@ -0,0 +1,518 @@
+package flow
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"slices"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/skupperproject/skupper/pkg/network"
+	"github.com/skupperproject/skupper/pkg/vanflow"
+	"github.com/skupperproject/skupper/pkg/vanflow/eventsource"
+	"github.com/skupperproject/skupper/pkg/vanflow/session"
+	"github.com/skupperproject/skupper/pkg/vanflow/store"
+	"golang.org/x/time/rate"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	v1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/util/retry"
+)
+
+var recordTypes []vanflow.Record = []vanflow.Record{
+	vanflow.SiteRecord{},
+	vanflow.RouterRecord{},
+	vanflow.LinkRecord{},
+	vanflow.RouterAccessRecord{},
+	vanflow.ConnectorRecord{},
+	vanflow.ListenerRecord{},
+	vanflow.ProcessRecord{},
+}
+
+type StatusSync struct {
+	records       store.Interface
+	recordMapping eventsource.RecordStoreMap
+
+	session       session.Container
+	discovery     *eventsource.Discovery
+	client        v1.ConfigMapInterface
+	configMapName string
+
+	logger *slog.Logger
+	ctx    context.Context
+
+	mu      sync.Mutex
+	clients map[string]*eventsource.Client
+
+	hasNext    chan struct{}
+	purgeQueue chan store.SourceRef
+	limit      rate.Limit
+	burst      int
+}
+
+func NewStatusSync(factory session.ContainerFactory, localSources map[string][]store.Interface, client v1.ConfigMapInterface, configMap string) *StatusSync {
+
+	// TODO ignore local sources and use their stores
+	logger := slog.New(slog.Default().Handler()).With(
+		slog.String("component", "kube.flow.statusSync"),
+	)
+	sessionCtr := factory.Create()
+	sessionCtr.OnSessionError(func(err error) {
+		logger.Error("session error on discovery container", slog.Any("error", err))
+	})
+	discovery := eventsource.NewDiscovery(sessionCtr, eventsource.DiscoveryOptions{})
+
+	s := &StatusSync{
+		session:       sessionCtr,
+		discovery:     discovery,
+		configMapName: configMap,
+		client:        client,
+
+		recordMapping: make(eventsource.RecordStoreMap, len(recordTypes)),
+		clients:       make(map[string]*eventsource.Client),
+		purgeQueue:    make(chan store.SourceRef, 8),
+		hasNext:       make(chan struct{}, 1),
+		limit:         rate.Every(time.Second),
+		burst:         1,
+
+		logger: logger,
+	}
+	recordsStore := store.NewSyncMapStore(store.SyncMapStoreConfig{
+		Handlers: store.EventHandlerFuncs{
+			OnAdd:    s.recordAdded,
+			OnChange: s.recordChanged,
+			OnDelete: s.recordDeleted,
+		},
+		Indexers: map[string]store.Indexer{
+			store.SourceIndex: store.SourceIndexer,
+			store.TypeIndex:   store.TypeIndexer,
+			byTypeParent:      indexByTypeParent,
+			byAddress:         indexByAddress,
+			byParentHost:      indexByParentHost,
+		},
+	})
+
+	for _, record := range recordTypes {
+		s.recordMapping[record.GetTypeMeta().String()] = recordsStore
+	}
+
+	s.records = recordsStore
+
+	return s
+}
+
+func (s *StatusSync) Run(ctx context.Context) {
+	s.session.Start(ctx)
+	s.ctx = ctx
+	discoveryErr := make(chan error, 1)
+	go func() {
+		discoveryErr <- s.discovery.Run(ctx, eventsource.DiscoveryHandlers{
+			Discovered: s.handleDiscovery,
+			Forgotten:  s.handleForgotten,
+		})
+	}()
+
+	var (
+		prev  network.NetworkStatusInfo
+		later <-chan time.Time
+	)
+
+	tryBuildAndPublish := func() {
+		var err error
+		prev, err = s.buildAndPublish(prev)
+		if err != nil {
+			s.logger.Error("could not update network status info", slog.Any("error", err))
+		}
+	}
+	var limiter *rate.Limiter
+	if s.limit != 0 && s.burst != 0 {
+		limiter = rate.NewLimiter(s.limit, s.burst)
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case err := <-discoveryErr:
+			if ctx.Err() != nil {
+				return
+			}
+			s.logger.Error("event source discovery finished unexpectedly", slog.Any("error", err))
+
+		case <-s.hasNext:
+			if limiter != nil {
+				r := limiter.Reserve()
+				if d := r.Delay(); d != 0 {
+					// rate limit reached
+					if later == nil {
+						later = time.After(d) // schedule next update
+					} else {
+						r.Cancel() // future update already scheduled
+					}
+					continue
+				}
+			}
+			tryBuildAndPublish()
+
+		case <-later:
+			later = nil
+			tryBuildAndPublish()
+		case source := <-s.purgeQueue:
+			ct := s.purge(source)
+			s.logger.Info("purged records from forgotten source",
+				slog.String("source", source.ID),
+				slog.Int("count", ct),
+			)
+		}
+	}
+}
+
+func (s *StatusSync) buildAndPublish(prev network.NetworkStatusInfo) (network.NetworkStatusInfo, error) {
+	next := s.build()
+	if cmp.Equal(prev, next) {
+		s.logger.Debug("no change since last publish")
+		return prev, nil
+	}
+	s.logger.Info("updating network status info", slog.String("configmap", s.configMapName))
+	return next, s.publish(next)
+}
+
+func (s *StatusSync) build() network.NetworkStatusInfo {
+	var (
+		info         network.NetworkStatusInfo
+		siteExemplar = store.Entry{Record: vanflow.SiteRecord{}}
+	)
+
+	sorted := func(entries []store.Entry) []store.Entry {
+		slices.SortFunc(entries, func(a, b store.Entry) int {
+			return strings.Compare(a.Record.Identity(), b.Record.Identity())
+		})
+		return entries
+	}
+
+	siteEntries := sorted(s.records.Index(store.TypeIndex, siteExemplar))
+	for _, entry := range siteEntries {
+		var siteInfo network.SiteStatusInfo
+		site := entry.Record.(vanflow.SiteRecord)
+		siteID := entry.Record.Identity()
+		siteInfo.Site = asSiteInfo(site)
+
+		exemplar := store.Entry{Record: vanflow.RouterRecord{
+			Parent: &siteID,
+		}}
+		routerEntries := sorted(s.records.Index(byTypeParent, exemplar))
+		for _, routerEnt := range routerEntries {
+			routerID := routerEnt.Record.Identity()
+			routerRecord := routerEnt.Record.(vanflow.RouterRecord)
+			var routerInfo network.RouterStatusInfo
+			routerInfo.Router = asRouterInfo(routerRecord)
+
+			linkExemplar := store.Entry{Record: vanflow.LinkRecord{
+				Parent: &routerID,
+			}}
+			linkEntries := sorted(s.records.Index(byTypeParent, linkExemplar))
+			routerInfo.Links = make([]network.LinkInfo, len(linkEntries))
+			for i, linkEnt := range linkEntries {
+				routerInfo.Links[i] = asLinkInfo(linkEnt.Record.(vanflow.LinkRecord))
+			}
+			connExemplar := store.Entry{Record: vanflow.ConnectorRecord{
+				Parent: &routerID,
+			}}
+			connEntries := sorted(s.records.Index(byTypeParent, connExemplar))
+			routerInfo.Connectors = make([]network.ConnectorInfo, len(connEntries))
+			for i, connEnt := range connEntries {
+				connector := connEnt.Record.(vanflow.ConnectorRecord)
+				connectorInfo := asConnectorInfo(connector)
+
+				var matchedProc bool
+				if connector.ProcessID != nil {
+					if procEntry, ok := s.records.Get(*connector.ProcessID); ok {
+						matchedProc = true
+						proc := procEntry.Record.(vanflow.ProcessRecord)
+						if proc.Name != nil {
+							connectorInfo.Target = *proc.Name
+						}
+					}
+				}
+				if !matchedProc {
+					procExemplar := store.Entry{Record: vanflow.ProcessRecord{
+						Parent:     &siteID,
+						SourceHost: connector.DestHost,
+					}}
+					procs := s.records.Index(byParentHost, procExemplar)
+					if len(procs) > 0 {
+						first := procs[0].Record.(vanflow.ProcessRecord)
+						connectorInfo.Process = first.ID
+						if first.Name != nil {
+							connectorInfo.Target = *first.Name
+						}
+					}
+				}
+				routerInfo.Connectors[i] = connectorInfo
+			}
+
+			lstExemplar := store.Entry{Record: vanflow.ListenerRecord{
+				Parent: &routerID,
+			}}
+			lstEntries := sorted(s.records.Index(byTypeParent, lstExemplar))
+			routerInfo.Listeners = make([]network.ListenerInfo, len(lstEntries))
+			for i, lstEnt := range lstEntries {
+				routerInfo.Listeners[i] = asListenerInfo(lstEnt.Record.(vanflow.ListenerRecord))
+			}
+
+			siteInfo.RouterStatus = append(siteInfo.RouterStatus, routerInfo)
+		}
+
+		info.SiteStatus = append(info.SiteStatus, siteInfo)
+	}
+
+	addresses := s.records.IndexValues(byAddress)
+	slices.Sort(addresses)
+	for _, address := range addresses {
+		var (
+			connectors []vanflow.ConnectorRecord
+			listeners  []vanflow.ListenerRecord
+		)
+		entries := sorted(s.records.Index(byAddress, store.Entry{
+			Record: vanflow.ConnectorRecord{Address: &address},
+		}))
+
+		for _, entry := range entries {
+			switch record := entry.Record.(type) {
+			case vanflow.ConnectorRecord:
+				connectors = append(connectors, record)
+			case vanflow.ListenerRecord:
+				listeners = append(listeners, record)
+			}
+		}
+
+		info.Addresses = append(info.Addresses, asAddressInfo(address, connectors, listeners))
+	}
+
+	return info
+}
+
+func (s *StatusSync) publish(info network.NetworkStatusInfo) error {
+	ctx, cancel := context.WithTimeout(s.ctx, time.Second*10)
+	defer cancel()
+	bs, err := json.Marshal(info)
+	if err != nil {
+		return fmt.Errorf("failed to marshal network info: %s", err)
+	}
+	networkStatus := string(bs)
+	data := map[string]string{"NetworkStatus": networkStatus}
+	err = retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		current, err := s.client.Get(ctx, s.configMapName, metav1.GetOptions{})
+		if err != nil {
+			return err
+		}
+		current.Data = data
+		_, err = s.client.Update(ctx, current, metav1.UpdateOptions{})
+		return err
+	})
+	if err != nil {
+		return fmt.Errorf("failed to update configmap: %s", err)
+	}
+	return nil
+}
+
+func (s *StatusSync) recordAdded(entry store.Entry) {
+	s.recordChanged(store.Entry{}, entry)
+}
+
+func (s *StatusSync) recordChanged(prev, entry store.Entry) {
+	switch record := entry.Record.(type) {
+	case vanflow.SiteRecord:
+		if isEndTimeSet(record.EndTime) {
+			s.records.Delete(record.Identity())
+		}
+	case vanflow.RouterRecord:
+		if isEndTimeSet(record.EndTime) {
+			s.records.Delete(record.Identity())
+		}
+	case vanflow.LinkRecord:
+		if isEndTimeSet(record.EndTime) {
+			s.records.Delete(record.Identity())
+		}
+	case vanflow.RouterAccessRecord:
+		if isEndTimeSet(record.EndTime) {
+			s.records.Delete(record.Identity())
+		}
+	case vanflow.ConnectorRecord:
+		if isEndTimeSet(record.EndTime) {
+			s.records.Delete(record.Identity())
+		}
+	case vanflow.ListenerRecord:
+		if isEndTimeSet(record.EndTime) {
+			s.records.Delete(record.Identity())
+		}
+	case vanflow.ProcessRecord:
+		if isEndTimeSet(record.EndTime) {
+			s.records.Delete(record.Identity())
+		}
+	}
+
+	s.notify()
+}
+func (s *StatusSync) recordDeleted(entry store.Entry) {
+	s.notify()
+}
+
+func (s *StatusSync) notify() {
+	select {
+	case s.hasNext <- struct{}{}:
+	default:
+	}
+}
+
+func (s *StatusSync) purge(source store.SourceRef) int {
+	matching := s.records.Index(store.SourceIndex, store.Entry{Metadata: store.Metadata{Source: source}})
+	for _, record := range matching {
+		s.records.Delete(record.Record.Identity())
+	}
+	return len(matching)
+}
+
+func (s *StatusSync) handleDiscovery(source eventsource.Info) {
+	client := eventsource.NewClient(s.session, eventsource.ClientOptions{
+		Source: source,
+	})
+
+	// register client with discovery to update lastseen, and monitor for staleness
+	err := s.discovery.NewWatchClient(s.ctx, eventsource.WatchConfig{
+		Client:      client,
+		ID:          source.ID,
+		Timeout:     time.Second * 30,
+		GracePeriod: time.Second * 30,
+	})
+
+	if err != nil {
+		s.logger.Error("error creating watcher for discoverd source", slog.Any("error", err))
+		s.discovery.Forget(source.ID)
+		return
+	}
+
+	router := eventsource.RecordStoreRouter{
+		Stores: s.recordMapping,
+		Source: sourceRef(source),
+	}
+	client.OnRecord(router.Route)
+	client.Listen(s.ctx, eventsource.FromSourceAddress())
+	if source.Type == "CONTROLLER" {
+		client.Listen(s.ctx, eventsource.FromSourceAddressHeartbeats())
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.clients[source.ID] = client
+
+	go func() {
+		ctx, cancel := context.WithTimeout(s.ctx, time.Second*5)
+		defer cancel()
+		if err := eventsource.FlushOnFirstMessage(ctx, client); err != nil {
+			if errors.Is(err, ctx.Err()) {
+				s.logger.Info("timed out waiting for first message. sending flush anyways")
+				err = client.SendFlush(s.ctx)
+			}
+			if err != nil {
+				s.logger.Error("error sending flush", slog.Any("error", err))
+			}
+		}
+	}()
+}
+
+func (s *StatusSync) handleForgotten(source eventsource.Info) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	client, ok := s.clients[source.ID]
+	if ok {
+		client.Close()
+		delete(s.clients, source.ID)
+	}
+	s.purgeQueue <- sourceRef(source)
+}
+
+func sourceRef(source eventsource.Info) store.SourceRef {
+	return store.SourceRef{
+		Version: fmt.Sprint(source.Version),
+		ID:      source.ID,
+	}
+}
+
+func isEndTimeSet(endtime *vanflow.Time) bool {
+	return endtime != nil && endtime.After(time.Unix(1, 0))
+}
+
+func asAddressInfo(address string, connectors []vanflow.ConnectorRecord, listeners []vanflow.ListenerRecord) network.AddressInfo {
+	var addressInfo network.AddressInfo
+	addressInfo.Name = address
+	for _, connector := range connectors {
+		if connector.Protocol != nil {
+			addressInfo.Protocol = *connector.Protocol
+			break
+		}
+	}
+	addressInfo.ConnectorCount = len(connectors)
+	addressInfo.ListenerCount = len(listeners)
+	return addressInfo
+}
+
+func asLinkInfo(link vanflow.LinkRecord) network.LinkInfo {
+	return network.LinkInfo{
+		Name:     dref(link.Name),
+		LinkCost: dref(link.LinkCost),
+	}
+}
+
+func asConnectorInfo(connector vanflow.ConnectorRecord) network.ConnectorInfo {
+	return network.ConnectorInfo{
+		DestHost: dref(connector.DestHost),
+		DestPort: dref(connector.DestPort),
+		Address:  dref(connector.Address),
+		Process:  dref(connector.ProcessID),
+	}
+}
+
+func asListenerInfo(listener vanflow.ListenerRecord) network.ListenerInfo {
+	return network.ListenerInfo{
+		//TODO(ck) Name not in spec? Name: dref(listener.Name),
+		DestHost: dref(listener.DestHost),
+		DestPort: dref(listener.DestPort),
+		Protocol: dref(listener.Protocol),
+		Address:  dref(listener.Address),
+	}
+}
+
+func asSiteInfo(site vanflow.SiteRecord) network.SiteInfo {
+	return network.SiteInfo{
+		Identity:  site.ID,
+		Name:      dref(site.Name),
+		Namespace: dref(site.Namespace),
+		Platform:  dref(site.Platform),
+		Version:   dref(site.Version),
+		Policy:    dref(site.Policy),
+	}
+}
+
+func asRouterInfo(router vanflow.RouterRecord) network.RouterInfo {
+	return network.RouterInfo{
+		Name:         dref(router.Name),
+		Namespace:    dref(router.Namespace),
+		Mode:         dref(router.Mode),
+		ImageName:    dref(router.ImageName),
+		ImageVersion: dref(router.ImageVersion),
+		Hostname:     dref(router.Hostname),
+	}
+}
+
+func dref[T any, R *T](ptr R) T {
+	var out T
+	if ptr != nil {
+		out = *ptr
+	}
+	return out
+}
diff --git a/pkg/vanflow/base.go b/pkg/vanflow/base.go
new file mode 100644
index 00000000..6f4491be
--- /dev/null
+++ b/pkg/vanflow/base.go
@@ -0,0 +1,77 @@
+package vanflow
+
+import (
+	"fmt"
+	"math"
+	"time"
+
+	"github.com/skupperproject/skupper/pkg/vanflow/encoding"
+)
+
+type TypeMeta struct {
+	APIVersion string
+	Type       string
+}
+
+func (m TypeMeta) String() string {
+	return fmt.Sprintf("%s/%s", m.APIVersion, m.Type)
+}
+
+type Record interface {
+	Identity() string
+	GetTypeMeta() TypeMeta
+}
+
+type BaseRecord struct {
+	ID        string `vflow:"1,required"`
+	StartTime *Time  `vflow:"3"`
+	EndTime   *Time  `vflow:"4"`
+}
+
+func NewBase(id string, times ...time.Time) BaseRecord {
+	base := BaseRecord{ID: id}
+	nextTime := func() *Time {
+		if len(times) > 0 {
+			var first Time
+			first, times = Time{times[0]}, times[1:]
+			return &first
+		}
+		return nil
+	}
+	base.StartTime = nextTime()
+	base.EndTime = nextTime()
+	if add := nextTime(); add != nil {
+		panic("expected at most two times for start and end")
+	}
+	return base
+}
+func (b BaseRecord) Identity() string {
+	return b.ID
+}
+
+type Time struct {
+	time.Time
+}
+
+func (t Time) EncodeRecordAttribute() (any, error) {
+	if t.Time.IsZero() {
+		return nil, encoding.ErrAttributeNotSet
+	}
+	ts := t.Time.UnixMicro()
+	if ts < 0 {
+		return nil, fmt.Errorf("cannot represent times before epoch in this encoding: %d", ts)
+	}
+	return uint64(ts), nil
+}
+
+func (t *Time) DecodeRecordAttribute(attr any) error {
+	uintAttr, ok := attr.(uint64)
+	if !ok {
+		return fmt.Errorf("expected type uint64 for timestamp but got %T", attr)
+	}
+	if uintAttr > math.MaxInt64 { // undefined past the year 294246
+		return fmt.Errorf("time too far in future for internal representation: %#x", uintAttr)
+	}
+	t.Time = time.UnixMicro(int64(uintAttr))
+	return nil
+}
diff --git a/pkg/vanflow/doc.go b/pkg/vanflow/doc.go
new file mode 100644
index 00000000..514aabd8
--- /dev/null
+++ b/pkg/vanflow/doc.go
@@ -0,0 +1,5 @@
+/*
+Package vanflow defines types for the message and record types specified in the
+VanFlow specification.
+*/
+package vanflow
diff --git a/pkg/vanflow/encoding/decoder.go b/pkg/vanflow/encoding/decoder.go
new file mode 100644
index 00000000..681f4e35
--- /dev/null
+++ b/pkg/vanflow/encoding/decoder.go
@@ -0,0 +1,122 @@
+package encoding
+
+import (
+	"errors"
+	"fmt"
+	"reflect"
+)
+
+var (
+	recordAttributeDecoderType = reflect.TypeOf((*RecordAttributeDecoder)(nil)).Elem()
+)
+
+// RecordAttributeDecoder is implemented by record attribute types that can
+// handle their own decoding.
+type RecordAttributeDecoder interface {
+	DecodeRecordAttribute(attr interface{}) error
+}
+
+// Decode a record from its record attribute set. Decode is only aware of
+// record types registered with MustRegisterRecord. Input must be a map type
+// containing uint32 keys (map[uint32]X or map[any]X)
+func Decode(recordset RecordAttributeSet) (interface{}, error) {
+	if recordset == nil {
+		return nil, errors.New("decode error: cannot decode nil record attribute set")
+	}
+	recordTypeCode, ok := recordset[typeOfRecord]
+	if !ok {
+		return nil, errors.New("decode error: record type attribute not present")
+	}
+
+	codepoint, ok := recordTypeCode.(uint32)
+	if !ok {
+		return nil, fmt.Errorf("decode error: unexpected type for record type attribute \"%T\"", recordTypeCode)
+	}
+
+	mu.RLock()
+	defer mu.RUnlock()
+	encoding, ok := decodings[codepoint]
+	if !ok {
+		return nil, fmt.Errorf("decode error: unknown record type for %d", recordTypeCode)
+	}
+	recordV := reflect.New(encoding.t)
+	if err := encoding.decode(recordset, recordV); err != nil {
+		return nil, fmt.Errorf("decode error: %w", err)
+	}
+	return recordV.Elem().Interface(), nil
+}
+
+type fieldDecoder interface {
+	decode(obj interface{}, v reflect.Value) error
+}
+
+func getFieldDecoder(t reflect.Type) (fieldDecoder, error) {
+	if t.Kind() != reflect.Pointer && reflect.PointerTo(t).Implements(recordAttributeDecoderType) {
+		return recordAttrFieldDecoder{}, nil
+	}
+	if t.Implements(recordAttributeDecoderType) {
+		return recordAttrFieldDecoder{}, nil
+	}
+	switch t.Kind() {
+	case reflect.Pointer:
+		child, err := getFieldDecoder(t.Elem())
+		return pointerDecoder{child}, err
+	case reflect.String, reflect.Uint64, reflect.Int64, reflect.Uint32, reflect.Int32:
+		return rawFieldDecoder{}, nil
+	case reflect.Struct:
+		fallthrough
+	default:
+		return nil, fmt.Errorf("unsupported attribute type %q", t)
+	}
+}
+
+type recordAttrFieldDecoder struct {
+}
+
+func (d recordAttrFieldDecoder) decode(attr interface{}, v reflect.Value) error {
+	if v.Kind() == reflect.Pointer && v.IsNil() {
+		v.Set(reflect.New(v.Type().Elem()))
+	}
+	if v.Kind() != reflect.Pointer &&
+		v.CanAddr() &&
+		reflect.PointerTo(v.Type()).Implements(recordAttributeDecoderType) {
+		v = v.Addr()
+	}
+	obj := v.Interface()
+	decoder, ok := obj.(RecordAttributeDecoder)
+	if !ok {
+		panic(fmt.Sprintf("expected %T to implement RecordAttributeUnmarshaler", obj))
+	}
+	return decoder.DecodeRecordAttribute(attr)
+}
+
+type rawFieldDecoder struct{}
+
+func (d rawFieldDecoder) decode(attr interface{}, v reflect.Value) error {
+	if !v.CanSet() {
+		return fmt.Errorf("field not addressable")
+	}
+	attrV := reflect.ValueOf(attr)
+	if !v.Type().AssignableTo(attrV.Type()) {
+		return fmt.Errorf("cannot assign value to type")
+	}
+	v.Set(attrV)
+	return nil
+}
+
+type pointerDecoder struct {
+	sub fieldDecoder
+}
+
+func (d pointerDecoder) decode(attr interface{}, v reflect.Value) error {
+	if !v.CanSet() {
+		return fmt.Errorf("field not addressable")
+	}
+	if v.IsNil() {
+		v.Set(reflect.New(v.Type().Elem()))
+	}
+	if err := d.sub.decode(attr, v.Elem()); err != nil {
+		return err
+	}
+	return nil
+}
diff --git a/pkg/vanflow/encoding/encoder.go b/pkg/vanflow/encoding/encoder.go
new file mode 100644
index 00000000..c2e58bdb
--- /dev/null
+++ b/pkg/vanflow/encoding/encoder.go
@@ -0,0 +1,117 @@
+package encoding
+
+import (
+	"errors"
+	"fmt"
+	"reflect"
+)
+
+type encodingError string
+
+func (err encodingError) Error() string {
+	return string(err)
+}
+
+const (
+	// ErrAttributeNotSet can be returned from a RecordAttributeEncoder
+	// when the field should not be set
+	ErrAttributeNotSet = encodingError("record attribute not set")
+)
+
+var (
+	recordAttributeEncoderType = reflect.TypeOf((*RecordAttributeEncoder)(nil)).Elem()
+)
+
+type RecordAttributeSet = map[interface{}]interface{}
+
+// RecordAttributeEncoder is implemented by record attribute types to overwrite
+// the default encoding behavior
+type RecordAttributeEncoder interface {
+	EncodeRecordAttribute() (interface{}, error)
+}
+
+// Encode a record into a record attribute set so that it can be sent over the
+// vanflow protocol. Only records types that have been registered with
+// MustRegisterRecord can be encoded.
+func Encode(record any) (RecordAttributeSet, error) {
+	mu.RLock()
+	defer mu.RUnlock()
+	if record == nil {
+		return nil, errors.New("cannot encode nil record")
+	}
+	recordV := reflect.ValueOf(record)
+	recordT := recordV.Type()
+	encoding, ok := encodings[recordT]
+	if !ok {
+		if recordT.Kind() == reflect.Pointer {
+			recordV = recordV.Elem()
+			recordT = recordV.Type()
+			encoding, ok = encodings[recordT]
+		}
+		if !ok {
+			return nil, fmt.Errorf("encode error: unregistered record type %T", record)
+		}
+	}
+	result, err := encoding.encode(recordV)
+	if result != nil {
+		result[typeOfRecord] = encoding.codepoint
+	}
+	return result, err
+}
+
+type fieldEncoder interface {
+	encode(v reflect.Value) (interface{}, error)
+}
+
+func getFieldEncoder(t reflect.Type) (fieldEncoder, error) {
+	if t.Implements(recordAttributeEncoderType) {
+		return recordAttrFieldEncoder{}, nil
+	}
+	switch t.Kind() {
+	case reflect.Pointer:
+		child, err := getFieldEncoder(t.Elem())
+		return pointerEncoder{child}, err
+	case reflect.String, reflect.Uint64, reflect.Int64, reflect.Uint32, reflect.Int32:
+		return rawFieldEncoder{}, nil
+	case reflect.Struct:
+		fallthrough
+	default:
+		return nil, fmt.Errorf("unsupported attribute type %q", t)
+	}
+}
+
+type rawFieldEncoder struct {
+}
+
+func (e rawFieldEncoder) encode(v reflect.Value) (interface{}, error) {
+	if v.IsZero() {
+		return nil, ErrAttributeNotSet
+	}
+	return v.Interface(), nil
+}
+
+type pointerEncoder struct {
+	sub fieldEncoder
+}
+
+func (e pointerEncoder) encode(v reflect.Value) (interface{}, error) {
+	if v.IsNil() {
+		return nil, ErrAttributeNotSet
+	}
+	return e.sub.encode(v.Elem())
+}
+
+type recordAttrFieldEncoder struct {
+}
+
+func (e recordAttrFieldEncoder) encode(v reflect.Value) (interface{}, error) {
+	if v.Kind() == reflect.Pointer && v.IsNil() {
+		return nil, ErrAttributeNotSet
+	}
+	val := v.Interface()
+	encoder, ok := val.(RecordAttributeEncoder)
+	if !ok {
+		panic(fmt.Sprintf("encoder error: type %s does not implement RecordAttributeEncoder", v.Type()))
+	}
+	return encoder.EncodeRecordAttribute()
+}
diff --git a/pkg/vanflow/encoding/encoding.go b/pkg/vanflow/encoding/encoding.go
new file mode 100644
index 00000000..24ee817b
--- /dev/null
+++ b/pkg/vanflow/encoding/encoding.go
@@ -0,0 +1,218 @@
+// encoding exposes functionality to convert between an arbitray maps and
+// native Go structs in service of the Vanflow protocol.
+//
+// The Vanflow protocol uses the amqp-value section to represent record types
+// in map structures. Typically go amqp clients perform the marshaling to and
+// from byte representation to some arbitrary map types, usually map[any]any.
+// This package provides mapping to and from that map type to Go types.
+//
+// All vanflow record types have an associated record type codepoint. Types
+// need to be registered with that codepoint using MustRegisterRecord before
+// they can be used.
+//
+// When decoding to a struct, encoding will look for `vflow` struct fields to
+// establish a mapping between vanflow record attribute codepoints and struct
+// fields. If a record attribute is required as part of the record, appending
+// ",required" to the vflow tag value will enforce that the field is set.
+//
+// For Example:
+//
+//	type ExampleRecord struct {
+//	  ExampleID int64 `vflow:"99,required"`
+//	  ExampleVal string `vflow:"100"`
+//	}
+//
+// The associated record attribute set would look like this:
+//
+//	map[any]any{
+//	  uint32(0): ExampleRecordCodepoint,
+//	  uint32(99): int64(1),
+//	  uint32(100): "test",
+//	}
+package encoding
+
+import (
+	"errors"
+	"fmt"
+	"reflect"
+	"strconv"
+	"strings"
+	"sync"
+)
+
+var (
+	mu        sync.RWMutex
+	encodings = map[reflect.Type]*typeEncoding{}
+	decodings = map[uint32]*typeEncoding{}
+)
+
+const (
+	vflowTag     = "vflow"
+	typeOfRecord = uint32(0)
+)
+
+type typeEncoding struct {
+	t         reflect.Type
+	codepoint uint32
+	fields    []field
+}
+
+// MustRegisterRecord registers a record type for Encoding/Decoding panics if
+// the type is not compatible, or if the codepoint or type has already been
+// registered.
+func MustRegisterRecord(codepoint uint32, record interface{}) {
+	mu.Lock()
+	defer mu.Unlock()
+	if _, ok := decodings[codepoint]; ok {
+		panic(fmt.Sprintf("cannot register record type %T using codepoint %d: already in use", record, codepoint))
+	}
+
+	encoding := newEncodingForType(codepoint, record)
+	if existing, ok := encodings[encoding.t]; ok {
+		panic(fmt.Sprintf("cannot register same type more than once. type %T already registered with code %d", record, existing.codepoint))
+	}
+	encodings[encoding.t] = encoding
+	decodings[codepoint] = encoding
+}
+
+func newEncodingForType(codepoint uint32, record interface{}) *typeEncoding {
+	recordType := reflect.Indirect(reflect.ValueOf(record)).Type()
+	encoding := typeEncoding{codepoint: codepoint, t: recordType}
+
+	if recordType.Kind() != reflect.Struct {
+		panic(fmt.Sprintf("unsupported encoder type %v. Expects struct type", recordType))
+	}
+
+	fields := fieldsForType(recordType, nil)
+	codepoints := map[uint32]field{}
+	for _, field := range fields {
+		if existing, ok := codepoints[field.Codepoint]; ok {
+			panic(fmt.Sprintf(
+				"struct field %s repeats vflow tag \"%d\" also used by %s",
+				field.Name, field.Codepoint, existing.Name))
+		}
+		codepoints[field.Codepoint] = field
+	}
+	encoding.fields = fields
+	return &encoding
+}
+
+type field struct {
+	Name      string
+	Codepoint uint32
+	Identity  bool
+
+	index   []int
+	encoder fieldEncoder
+	decoder fieldDecoder
+}
+
+func parseFieldOpts(f reflect.StructField) (field, bool) {
+	opts := field{
+		Name: f.Name,
+	}
+
+	tag := f.Tag.Get(vflowTag)
+	if tag == "" {
+		return opts, false
+	}
+	sCode, sOpt, _ := strings.Cut(tag, ",")
+	code, err := strconv.ParseUint(sCode, 10, 32)
+	if err != nil {
+		panic(fmt.Sprintf("vflow struct tag parse error for field %s: %s", f.Name, err))
+	}
+	opts.Codepoint = uint32(code)
+	switch suffix := sOpt; suffix {
+	case "required":
+		opts.Identity = true
+	case "":
+	default:
+		panic(fmt.Sprintf("vflow struct tag parse error: unexpected option %s", suffix))
+	}
+	return opts, true
+}
+
+func (e typeEncoding) encode(root reflect.Value) (RecordAttributeSet, error) {
+	attributeSet := make(RecordAttributeSet)
+	var v reflect.Value
+	for _, field := range e.fields {
+		v = root
+		for _, idx := range field.index {
+			v = v.Field(idx)
+		}
+		out, err := field.encoder.encode(v)
+		if err != nil {
+			if errors.Is(err, ErrAttributeNotSet) {
+				if field.Identity {
+					return attributeSet, fmt.Errorf("missing or empty required field %s: %s", field.Name, err)
+				}
+				continue // ignore if not required
+			}
+			return attributeSet, fmt.Errorf("error encoding field %q: %w", field.Name, err)
+		}
+		if out != nil {
+			attributeSet[field.Codepoint] = out
+		}
+	}
+	return attributeSet, nil
+}
+
+func (e typeEncoding) decode(attrs RecordAttributeSet, root reflect.Value) error {
+	var v reflect.Value
+	for _, field := range e.fields {
+		obj, ok := attrs[field.Codepoint]
+		if !ok {
+			if field.Identity {
+				return fmt.Errorf("record attribute set missing required field %q", field.Name)
+			}
+			continue
+		}
+		v = root.Elem()
+		for _, idx := range field.index {
+			v = v.Field(idx)
+		}
+		err := field.decoder.decode(obj, v)
+		if err != nil {
+			return fmt.Errorf("error decoding field %q: %w", field.Name, err)
+		}
+	}
+	return nil
+}
+
+func fieldsForType(t reflect.Type, index []int) []field {
+	var fields []field
+	for i := 0; i < t.NumField(); i++ {
+		f := t.Field(i)
+		next := make([]int, len(index)+1)
+		copy(next, index)
+		next[len(next)-1] = i
+		// Include fields from embedded structs
+		if f.Anonymous {
+			if f.Type.Kind() == reflect.Struct {
+				fields = append(fields, fieldsForType(f.Type, next)...)
+			}
+			continue
+		} else if !f.IsExported() {
+			// ignore unexported fields
+			continue
+		}
+
+		fieldOpts, ok := parseFieldOpts(f)
+		if !ok {
+			continue
+		}
+		fieldOpts.index = next
+		encoder, err := getFieldEncoder(f.Type)
+		if err != nil {
+			panic(fmt.Sprintf("invalid vflow field encoder for %q: %s", f.Name, err))
+		}
+		fieldOpts.encoder = encoder
+		decoder, err := getFieldDecoder(f.Type)
+		if err != nil {
+			panic(fmt.Sprintf("invalid vflow field decoder for %q: %s", f.Name, err))
+		}
+		fieldOpts.decoder = decoder
+		fields = append(fields, fieldOpts)
+	}
+	return fields
+}
diff --git a/pkg/vanflow/eventsource/client.go b/pkg/vanflow/eventsource/client.go
new file mode 100644
index 00000000..39d4743b
--- /dev/null
+++ b/pkg/vanflow/eventsource/client.go
@@ -0,0 +1,197 @@
+package eventsource
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"sync"
+
+	"github.com/skupperproject/skupper/pkg/vanflow"
+	"github.com/skupperproject/skupper/pkg/vanflow/session"
+)
+
+const (
+	sourceSuffixFlows      = ".flows"
+	sourceSuffixLogs       = ".logs"
+	sourceSuffixHeartbeats = ".heartbeats"
+)
+
+// Client is responsible for interacting with Event Sources
+//
+// Allows the caller to register message handler callbacks to react to received
+// messages and to control exactly what sources the client listens to.
+type Client struct {
+	container   session.Container
+	eventSource Info
+
+	lock              sync.Mutex
+	cleanup           []func()
+	heartbeatHandlers []HeartbeatMessageHandler
+	recordHandlers    []RecordMessageHandler
+
+	wg     sync.WaitGroup
+	logger *slog.Logger
+}
+
+type ClientOptions struct {
+	// Source of vanflow events
+	Source Info
+}
+
+func NewClient(container session.Container, cfg ClientOptions) *Client {
+	c := &Client{
+		container:   container,
+		eventSource: cfg.Source,
+		logger: slog.New(slog.Default().Handler()).With(
+			slog.String("component", "vanflow.eventsource.client"),
+			slog.String("instance", cfg.Source.ID),
+		),
+	}
+	c.start()
+	return c
+}
+
+func (c *Client) start() {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	done := make(chan struct{})
+	c.cleanup = append(c.cleanup, func() { close(done) })
+}
+
+type HeartbeatMessageHandler func(vanflow.HeartbeatMessage)
+type RecordMessageHandler func(vanflow.RecordMessage)
+
+// OnHeartbeat registers a callback handler for HeartbeatMessages.
+func (c *Client) OnHeartbeat(handler HeartbeatMessageHandler) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	c.heartbeatHandlers = append(c.heartbeatHandlers, handler)
+}
+
+// OnRecord registers a callback handler for RecordMessages.
+func (c *Client) OnRecord(handler RecordMessageHandler) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	c.recordHandlers = append(c.recordHandlers, handler)
+}
+
+func (c *Client) getHandlers() ([]RecordMessageHandler, []HeartbeatMessageHandler) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	records := make([]RecordMessageHandler, len(c.recordHandlers))
+	heartbeats := make([]HeartbeatMessageHandler, len(c.heartbeatHandlers))
+	copy(records, c.recordHandlers)
+	copy(heartbeats, c.heartbeatHandlers)
+	return records, heartbeats
+}
+
+// Listen instructs the Client to listen to an event source using the specified
+// listener configuration until the context is cancelled or client.Close() is
+// called.
+func (c *Client) Listen(ctx context.Context, attributes ListenerConfigProvider) error {
+	c.wg.Add(1)
+	listenerCtx, listenerCancel := context.WithCancel(ctx)
+
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	c.cleanup = append(c.cleanup, listenerCancel)
+
+	go func(ctx context.Context) {
+		defer c.wg.Done()
+		cfg := attributes.Get(c.eventSource)
+		receiver := c.container.NewReceiver(cfg.Address, session.ReceiverOptions{
+			Credit: cfg.Credit,
+		})
+		defer receiver.Close(ctx)
+		for {
+			amqpMsg, err := receiver.Next(ctx)
+			if err != nil {
+				if errors.Is(err, ctx.Err()) {
+					return
+				}
+				c.logger.Error("client error receiving message", slog.Any("error", err), slog.String("address", cfg.Address))
+				continue
+			}
+			if err := receiver.Accept(ctx, amqpMsg); err != nil {
+				if errors.Is(err, ctx.Err()) {
+					return
+				}
+				c.logger.Error("client error accepting message", slog.Any("error", err), slog.String("address", cfg.Address))
+				continue
+			}
+			decoded, err := vanflow.Decode(amqpMsg)
+			if err != nil {
+				c.logger.Error("skipping message that could not be decoded", slog.Any("error", err))
+				continue
+			}
+			recordHandlers, heartbeatHandlers := c.getHandlers()
+			switch message := decoded.(type) {
+			case vanflow.RecordMessage:
+				for _, handler := range recordHandlers {
+					handler(message)
+				}
+			case vanflow.HeartbeatMessage:
+				for _, handler := range heartbeatHandlers {
+					handler(message)
+				}
+			}
+		}
+	}(listenerCtx)
+	return nil
+}
+
+// Close stops all listeners
+func (c *Client) Close() {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	for _, cancel := range c.cleanup {
+		cancel()
+	}
+	c.wg.Wait()
+}
+
+// SendFlush sends a FlushMessage to the Event Source
+func (c *Client) SendFlush(ctx context.Context) error {
+	var flush vanflow.FlushMessage
+	flush.To = c.eventSource.Direct
+	msg := flush.Encode()
+
+	sender := c.container.NewSender(c.eventSource.Direct, session.SenderOptions{})
+	defer sender.Close(ctx)
+	if err := sender.Send(ctx, msg); err != nil {
+		return fmt.Errorf("client flush error: %w", err)
+	}
+	return nil
+}
+
+type ListenerConfig struct {
+	Address string
+	Credit  int
+}
+
+type ListenerConfigProvider interface {
+	Get(Info) ListenerConfig
+}
+
+type addresser func(i Info) string
+
+func (fn addresser) Get(info Info) ListenerConfig {
+	return ListenerConfig{Address: fn(info), Credit: 256}
+}
+
+func FromSourceAddress() ListenerConfigProvider {
+	return addresser(func(i Info) string { return i.Address })
+}
+
+func FromSourceAddressLogs() ListenerConfigProvider {
+	return addresser(func(i Info) string { return i.Address + sourceSuffixLogs })
+}
+
+func FromSourceAddressFlows() ListenerConfigProvider {
+	return addresser(func(i Info) string { return i.Address + sourceSuffixFlows })
+}
+
+func FromSourceAddressHeartbeats() ListenerConfigProvider {
+	return addresser(func(i Info) string { return i.Address + sourceSuffixHeartbeats })
+}
diff --git a/pkg/vanflow/eventsource/discovery.go b/pkg/vanflow/eventsource/discovery.go
new file mode 100644
index 00000000..f0e24659
--- /dev/null
+++ b/pkg/vanflow/eventsource/discovery.go
@@ -0,0 +1,302 @@
+package eventsource
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/skupperproject/skupper/pkg/vanflow"
+	"github.com/skupperproject/skupper/pkg/vanflow/session"
+)
+
+const beaconAddress = "mc/sfe.all"
+
+// Discovery manages a collection of known event sources
+type Discovery struct {
+	container     session.Container
+	lock          sync.Mutex
+	state         map[string]Info
+	discovered    chan Info
+	forgotten     chan Info
+	beaconAddress string
+	logger        *slog.Logger
+}
+
+type DiscoveryHandlers struct {
+	Discovered func(source Info)
+	Forgotten  func(source Info)
+}
+
+type DiscoveryOptions struct {
+	BeaconAddress string
+}
+
+func NewDiscovery(container session.Container, opts DiscoveryOptions) *Discovery {
+	if opts.BeaconAddress == "" {
+		opts.BeaconAddress = beaconAddress
+	}
+	return &Discovery{
+		container:     container,
+		state:         make(map[string]Info),
+		discovered:    make(chan Info, 32),
+		forgotten:     make(chan Info, 32),
+		beaconAddress: opts.BeaconAddress,
+		logger: slog.New(slog.Default().Handler()).With(
+			slog.String("component", "vanflow.eventsource.discovery"),
+		),
+	}
+}
+
+// Run event source discovery until the context is cancelled
+func (d *Discovery) Run(ctx context.Context, handlers DiscoveryHandlers) error {
+	receiver := d.container.NewReceiver(d.beaconAddress, session.ReceiverOptions{})
+	defer receiver.Close(ctx)
+	dCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	go d.handleDiscovery(dCtx, handlers)
+	for {
+		msg, err := receiver.Next(ctx)
+		if err != nil {
+			return fmt.Errorf("discovery error receiving beacon messages: %w", err)
+		}
+		if err := receiver.Accept(ctx, msg); err != nil {
+			d.logger.Error("discovery got error accepting message", slog.Any("error", err))
+		}
+		if *msg.Properties.Subject != "BEACON" {
+			d.logger.Info("received non-beacon from beacon source", slog.Any("subject", msg.Properties.Subject), slog.Any("source", msg.Properties.To))
+			continue
+		}
+		d.observe(vanflow.DecodeBeacon(msg))
+	}
+}
+
+func (d *Discovery) handleDiscovery(ctx context.Context, handlers DiscoveryHandlers) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case info := <-d.discovered:
+			if handlers.Discovered != nil {
+				handlers.Discovered(info)
+			}
+		case info := <-d.forgotten:
+			if handlers.Forgotten != nil {
+				handlers.Forgotten(info)
+			}
+		}
+	}
+}
+
+// Get an EventSource by ID
+func (d *Discovery) Get(id string) (source Info, ok bool) {
+	d.lock.Lock()
+	defer d.lock.Unlock()
+	state, ok := d.state[id]
+	if !ok {
+		return source, false
+	}
+	return state, true
+}
+
+// Add an EventSource
+func (d *Discovery) Add(info Info) bool {
+	d.lock.Lock()
+	defer d.lock.Unlock()
+	if _, exists := d.state[info.ID]; exists {
+		return false
+	}
+	d.state[info.ID] = info
+	d.discovered <- info
+	return true
+}
+
+// Forget an EventSource
+func (d *Discovery) Forget(id string) bool {
+	var (
+		state  Info
+		forget bool
+	)
+	d.lock.Lock()
+	defer d.lock.Unlock()
+	state, forget = d.state[id]
+	delete(d.state, id)
+	if forget {
+		d.forgotten <- state
+	}
+	return forget
+}
+
+// List all known EventSources
+func (d *Discovery) List() []Info {
+	d.lock.Lock()
+	defer d.lock.Unlock()
+	results := make([]Info, 0, len(d.state))
+	for _, state := range d.state {
+		results = append(results, state)
+	}
+	return results
+}
+
+type WatchConfig struct {
+	// Client to watch
+	Client *Client
+	// ID of the event source to watch
+	ID string
+	// Timeout for client activity. When set, timeout is the duration that
+	// discovery will wait after the most recent client activity before
+	// forgetting the source.
+	Timeout time.Duration
+
+	// GracePeriod is the time to wait for client activity before enforcing
+	// the Timeout.
+	GracePeriod time.Duration
+
+	// DiscoveryUpdateInterval is the period at which discovery gets updated
+	// with the latest LastSeen timestamp from the watcher. Defaults to once
+	// per second.
+	DiscoveryUpdateInterval time.Duration
+}
+
+// NewWatchClient creates a client for a given event source and uses that client
+// to keep the source LastHeard time up to date.
+func (d *Discovery) NewWatchClient(ctx context.Context, cfg WatchConfig) error {
+	_, ok := d.Get(cfg.ID)
+	if !ok {
+		return fmt.Errorf("unknown event source %s", cfg.ID)
+	}
+
+	w := newWatch(d, cfg.Client)
+	go w.run(ctx, cfg)
+	return nil
+}
+
+type watch struct {
+	lastSeen  atomic.Pointer[time.Time]
+	discovery *Discovery
+	client    *Client
+}
+
+func newWatch(discovery *Discovery, client *Client) *watch {
+	w := watch{
+		discovery: discovery,
+		client:    client,
+	}
+	client.OnHeartbeat(func(vanflow.HeartbeatMessage) {
+		ts := time.Now()
+		w.lastSeen.Store(&ts)
+	})
+
+	client.OnRecord(func(vanflow.RecordMessage) {
+		ts := time.Now()
+		w.lastSeen.Store(&ts)
+	})
+	return &w
+}
+
+func (w *watch) run(ctx context.Context, cfg WatchConfig) {
+	var (
+		watchTimer  *time.Timer
+		watchTimerC <-chan time.Time
+	)
+
+	if cfg.Timeout > 0 {
+		watchTimer = time.NewTimer(cfg.GracePeriod + cfg.Timeout)
+		watchTimerC = watchTimer.C
+		defer watchTimer.Stop()
+	}
+
+	// only update discovery data once per second to keep
+	// lock contention reasonable.
+	advanceInterval := cfg.DiscoveryUpdateInterval
+	if advanceInterval <= 0 {
+		advanceInterval = time.Second
+	}
+	advanceTicker := time.NewTicker(advanceInterval)
+	defer advanceTicker.Stop()
+	defer w.client.Close()
+
+	var prevObserved time.Time
+	prevObserved, _ = w.advanceLastSeen(prevObserved)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-watchTimerC:
+			// Watch Timeout has been reached. Check one last time for watch
+			// activity before forgetting the event source.
+			next, ok := w.advanceLastSeen(prevObserved)
+			if !ok {
+				w.discovery.Forget(cfg.ID)
+				return
+			}
+			w.discovery.lastSeen(cfg.ID, next)
+			watchTimer.Reset(cfg.Timeout)
+			prevObserved = next
+		case <-advanceTicker.C:
+			next, ok := w.advanceLastSeen(prevObserved)
+			if !ok {
+				continue
+			}
+
+			w.discovery.lastSeen(cfg.ID, next)
+			if watchTimer != nil {
+				if !watchTimer.Stop() {
+					<-watchTimer.C
+				}
+				watchTimer.Reset(cfg.Timeout)
+			}
+			prevObserved = next
+		}
+	}
+}
+
+func (w *watch) advanceLastSeen(prev time.Time) (time.Time, bool) {
+	currentObserved := w.lastSeen.Load()
+	if currentObserved == nil || !currentObserved.After(prev) {
+		return prev, false
+	}
+	return *currentObserved, true
+}
+
+func (d *Discovery) observe(beacon vanflow.BeaconMessage) {
+	var (
+		state      Info
+		discovered bool
+	)
+	d.lock.Lock()
+	defer d.lock.Unlock()
+	tObserved := time.Now()
+	state, ok := d.state[beacon.Identity]
+	if !ok {
+		state = Info{
+			ID:       beacon.Identity,
+			Version:  int(beacon.Version),
+			Type:     beacon.SourceType,
+			Address:  beacon.Address,
+			Direct:   beacon.Direct,
+			LastSeen: tObserved,
+		}
+		discovered = true
+	}
+	state.LastSeen = tObserved
+	d.state[beacon.Identity] = state
+	if discovered {
+		d.discovered <- state
+	}
+}
+
+func (d *Discovery) lastSeen(id string, latest time.Time) {
+	d.lock.Lock()
+	defer d.lock.Unlock()
+	state, ok := d.state[id]
+	if !ok {
+		return
+	}
+	if latest.After(state.LastSeen) {
+		state.LastSeen = latest
+		d.state[id] = state
+	}
+}
diff --git a/pkg/vanflow/eventsource/doc.go b/pkg/vanflow/eventsource/doc.go
new file mode 100644
index 00000000..18b66ace
--- /dev/null
+++ b/pkg/vanflow/eventsource/doc.go
@@ -0,0 +1,5 @@
+/*
+The eventsource package contains mechanisims for discovering and interracting
+with vanflow event sources.
+*/
+package eventsource
diff --git a/pkg/vanflow/eventsource/flush.go b/pkg/vanflow/eventsource/flush.go
new file mode 100644
index 00000000..c0897189
--- /dev/null
+++ b/pkg/vanflow/eventsource/flush.go
@@ -0,0 +1,37 @@
+package eventsource
+
+import (
+	"context"
+	"sync"
+
+	"github.com/skupperproject/skupper/pkg/vanflow"
+)
+
+// FlushOnFirstMessage adds handlers to a client to listen for the first
+// heartbeat or record message received from the source and then calls
+// client.SendFlush().
+func FlushOnFirstMessage(ctx context.Context, client *Client) error {
+	done := make(chan struct{})
+	awaiter := messageAwaiter{done: done}
+	client.OnRecord(awaiter.Record)
+	client.OnHeartbeat(awaiter.Heartbeat)
+	select {
+	case <-done:
+		return client.SendFlush(ctx)
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+type messageAwaiter struct {
+	done chan struct{}
+	once sync.Once
+}
+
+func (f *messageAwaiter) Record(vanflow.RecordMessage) {
+	f.once.Do(func() { close(f.done) })
+}
+
+func (f *messageAwaiter) Heartbeat(vanflow.HeartbeatMessage) {
+	f.once.Do(func() { close(f.done) })
+}
diff --git a/pkg/vanflow/eventsource/manager.go b/pkg/vanflow/eventsource/manager.go
new file mode 100644
index 00000000..79a1c516
--- /dev/null
+++ b/pkg/vanflow/eventsource/manager.go
@@ -0,0 +1,358 @@
+package eventsource
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+	"time"
+
+	amqp "github.com/Azure/go-amqp"
+	"github.com/skupperproject/skupper/pkg/vanflow"
+	"github.com/skupperproject/skupper/pkg/vanflow/encoding"
+	"github.com/skupperproject/skupper/pkg/vanflow/session"
+	"github.com/skupperproject/skupper/pkg/vanflow/store"
+)
+
+type ManagerConfig struct {
+	Source Info
+
+	Stores []store.Interface
+
+	// HeartbeatInterval defaults to 2 seconds
+	HeartbeatInterval time.Duration
+	// BeaconInterval defaults to 10 seconds
+	BeaconInterval time.Duration
+
+	// UseAlternateHeartbeatAddress indicates that the manager should send
+	// heartbeat messages to the source address with the `.heartbeats` suffix.
+	UseAlternateHeartbeatAddress bool
+
+	// FlushDelay is the amount of time to wait after receiving an initial
+	// flush message before beginning to send updates.
+	FlushDelay time.Duration
+	// FlushBatchSize is the maximum number of records that will be sent in a
+	// single record message on flush.
+	FlushBatchSize int
+
+	// UpdateBufferTime is the amount of time to wait for a full batch of
+	// record updates before sending a partial batch.
+	UpdateBufferTime time.Duration
+	// UpdateBatchSize is the maximum number of record updates that will be
+	// sent in a single record message. Defaults to 1.
+	UpdateBatchSize int
+}
+
+type RecordUpdate struct {
+	Prev vanflow.Record
+	Curr vanflow.Record
+}
+
+type Manager struct {
+	ManagerConfig
+	container session.Container
+
+	flushQueue  chan struct{}
+	changeQueue chan RecordUpdate
+	sendQueue   chan vanflow.RecordMessage
+
+	logger *slog.Logger
+}
+
+func NewManager(container session.Container, cfg ManagerConfig) *Manager {
+	return &Manager{
+		container:     container,
+		ManagerConfig: cfg,
+		flushQueue:    make(chan struct{}, 8),
+		changeQueue:   make(chan RecordUpdate, 256),
+		sendQueue:     make(chan vanflow.RecordMessage, 256),
+		logger: slog.New(slog.Default().Handler()).With(
+			slog.String("component", "vanflow.eventsource.manager"),
+			slog.String("instance", cfg.Source.ID),
+		),
+	}
+}
+
+func (m *Manager) PublishUpdate(update RecordUpdate) {
+	m.changeQueue <- update
+}
+
+func (m *Manager) Run(ctx context.Context) {
+	go m.listenFlushes(ctx)
+	go m.sendKeepalives(ctx)
+	go m.sendRecords(ctx)
+	m.serve(ctx)
+}
+
+func (m *Manager) sendRecords(ctx context.Context) {
+	sender := m.container.NewSender(m.Source.Address, session.SenderOptions{})
+	defer sender.Close(ctx)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case record := <-m.sendQueue:
+			msg, err := record.Encode()
+			if err != nil {
+				m.logger.Error("skipping record message after encoding error:", slog.Any("error", err))
+				continue
+			}
+			if err := sender.Send(ctx, msg); err != nil {
+				m.logger.Error("error sending event source record", slog.Any("error", err))
+				continue
+			}
+			m.logger.Info("record message sent", slog.Int("record_count", len(record.Records)))
+		}
+	}
+}
+
+func (m *Manager) diffRecord(d RecordUpdate) (vanflow.Record, bool) {
+	if d.Prev == nil {
+		return d.Curr, true
+	}
+	prev, err := encoding.Encode(d.Prev)
+	if err != nil {
+		m.logger.Error("record update diff error encoding prev", slog.Any("error", err))
+		return nil, false
+	}
+	next, err := encoding.Encode(d.Curr)
+	if err != nil {
+		m.logger.Error("record update diff error encoding curr", slog.Any("error", err))
+		return nil, false
+	}
+	var isDiff bool
+	delta := make(map[any]any)
+	for k := range next {
+		pv, ok := prev[k]
+		if !ok {
+			isDiff = true
+			delta[k] = next[k]
+			continue
+		}
+		if next[k] != pv {
+			isDiff = true
+			delta[k] = next[k]
+		}
+		kCodePoint, ok := k.(uint32)
+		if ok && kCodePoint < 2 { // hack to keep identifying info in the record
+			delta[k] = next[k]
+		}
+	}
+	if !isDiff {
+		return nil, false
+	}
+	deltaRecord, err := encoding.Decode(delta)
+	if err != nil {
+		m.logger.Error("record update diff error decoding delta", slog.Any("error", err))
+		return nil, false
+	}
+	return deltaRecord.(vanflow.Record), true
+}
+
+func (m *Manager) serve(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case update := <-m.changeQueue:
+			var buffer []RecordUpdate
+			if m.UpdateBufferTime > 0 && m.UpdateBatchSize > 1 {
+				bufferCtx, cancelBuffer := context.WithTimeout(ctx, m.UpdateBufferTime)
+				buffer = nextN(bufferCtx, m.changeQueue, m.UpdateBatchSize-1)
+				cancelBuffer()
+				if ctx.Err() != nil {
+					return
+				}
+			}
+			buffer = append([]RecordUpdate{update}, buffer...)
+
+			var record vanflow.RecordMessage
+			record.Records = make([]vanflow.Record, 0, len(buffer))
+			for _, update := range buffer {
+				delta, changed := m.diffRecord(update)
+				if !changed {
+					continue
+				}
+				record.Records = append(record.Records, delta)
+			}
+			if len(record.Records) == 0 {
+				m.logger.Debug("record changes buffered but none were changed", slog.Int("record_count", len(buffer)))
+				continue
+			}
+			select {
+			case <-ctx.Done():
+				return
+			case m.sendQueue <- record:
+			}
+
+		case <-m.flushQueue:
+			// handle a flush
+			if m.FlushDelay > 0 {
+				drainCtx, cancelDrain := context.WithTimeout(ctx, m.FlushDelay)
+				nextN(drainCtx, m.flushQueue, 2048)
+				cancelDrain()
+				if ctx.Err() != nil {
+					return
+				}
+			}
+			m.logger.Info("servicing flush", slog.String("source", m.Source.ID))
+			for _, stor := range m.Stores {
+				entries := stor.List()
+
+				for len(entries) > 0 {
+					var batch []store.Entry
+					batch, entries = entries, entries[:0]
+					if len(batch) > m.FlushBatchSize && m.FlushBatchSize > 0 {
+						batch, entries = batch[:m.FlushBatchSize], batch[m.FlushBatchSize:]
+					}
+
+					var record vanflow.RecordMessage
+					record.Records = make([]vanflow.Record, len(batch))
+					for i, entry := range batch {
+						record.Records[i] = entry.Record
+					}
+					select {
+					case <-ctx.Done():
+						return
+					case m.sendQueue <- record:
+					}
+				}
+			}
+		}
+	}
+}
+
+func (m *Manager) sendKeepalives(ctx context.Context) {
+	beaconInterval := m.BeaconInterval
+	if beaconInterval <= 0 {
+		beaconInterval = 10 * time.Second
+	}
+	heartbeatInterval := m.HeartbeatInterval
+	if heartbeatInterval <= 0 {
+		heartbeatInterval = 2 * time.Second
+	}
+
+	beaconMessage := vanflow.BeaconMessage{
+		Version:    uint32(m.Source.Version),
+		SourceType: m.Source.Type,
+		Address:    m.Source.Address,
+		Direct:     m.Source.Direct,
+		Identity:   m.Source.ID,
+	}
+	beaconMessage.To = beaconAddress
+	heartbeatMessage := vanflow.HeartbeatMessage{
+		Version:  uint32(m.Source.Version),
+		Identity: m.Source.ID,
+	}
+
+	heartbeatAddr := m.Source.Address
+	if m.UseAlternateHeartbeatAddress {
+		heartbeatAddr = heartbeatAddr + sourceSuffixHeartbeats
+	}
+
+	beaconSender := m.container.NewSender(beaconAddress, session.SenderOptions{})
+	defer beaconSender.Close(ctx)
+	heartbeatSender := m.container.NewSender(heartbeatAddr, session.SenderOptions{})
+	defer heartbeatSender.Close(ctx)
+
+	// start with sending beacon
+	for {
+		if err := sendWithTimeout(ctx, beaconInterval, beaconSender, beaconMessage.Encode()); err != nil {
+			m.logger.Error("error sending initial beacon message", slog.Any("error", err))
+			continue
+		}
+		break
+	}
+
+	messageDeliveryTimeout := heartbeatInterval
+	if messageDeliveryTimeout > beaconInterval {
+		messageDeliveryTimeout = beaconInterval
+	}
+
+	beaconTimer := time.NewTicker(beaconInterval)
+	heartbeatTimer := time.NewTicker(heartbeatInterval)
+	heartbeatTimeouts := 0
+	firstHeartbeatSent := false
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-beaconTimer.C:
+			msg := beaconMessage.Encode()
+			if err := sendWithTimeout(ctx, messageDeliveryTimeout, beaconSender, msg); err != nil {
+				m.logger.Error("error sending event source beacon", slog.Any("error", err))
+			}
+		case <-heartbeatTimer.C:
+			heartbeatMessage.Now = uint64(time.Now().UnixMicro())
+			msg := heartbeatMessage.Encode()
+			// skupper router will block messages sent multicast without a
+			// listener by default. The router needs to register a beacon
+			// before heartbeats are unblocked. This doesn't seem entirely
+			// reliable so I've opted to ignore the first few heartbeat
+			// timeouts before aborting the connection entirely.
+			if err := sendWithTimeout(ctx, messageDeliveryTimeout, heartbeatSender, msg); err != nil {
+				if errors.Is(err, errSendTimeoutExceeded) && heartbeatTimeouts < 3 && !firstHeartbeatSent {
+					heartbeatTimeouts++
+					m.logger.Info("initial heartbeat message send timed out")
+					continue
+				}
+				m.logger.Error("error sending event source heartbeat",
+					slog.Any("error", err),
+					slog.Bool("priorSuccess", firstHeartbeatSent),
+					slog.Int("timeouts", heartbeatTimeouts))
+			}
+			firstHeartbeatSent = true
+		}
+	}
+}
+
+func (m *Manager) listenFlushes(ctx context.Context) {
+	flushReceiver := m.container.NewReceiver(m.Source.Direct, session.ReceiverOptions{Credit: 256})
+	defer flushReceiver.Close(ctx)
+	for {
+		msg, err := flushReceiver.Next(ctx)
+		if err != nil {
+			if errors.Is(err, ctx.Err()) {
+				return
+			}
+			m.logger.Error("flush receive error", slog.Any("error", err))
+		}
+		flushReceiver.Accept(ctx, msg)
+		select {
+		case m.flushQueue <- struct{}{}:
+		default: // drop flush if queue is full
+		}
+	}
+}
+
+var errSendTimeoutExceeded = errors.New("send timed out")
+
+func sendWithTimeout(ctx context.Context, timeout time.Duration, sender session.Sender, msg *amqp.Message) error {
+	requestCtx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+	err := sender.Send(requestCtx, msg)
+	if err != nil {
+		if errors.Is(err, requestCtx.Err()) {
+			return errSendTimeoutExceeded
+		}
+	}
+	return err
+}
+
+// pulls next N items from a channel
+func nextN[T any](ctx context.Context, c <-chan T, n int) []T {
+	var out []T
+	if n < 1 {
+		return out
+	}
+	for {
+		select {
+		case <-ctx.Done():
+			return out
+		case t := <-c:
+			out = append(out, t)
+			if len(out) == n {
+				return out
+			}
+		}
+	}
+}
diff --git a/pkg/vanflow/eventsource/routing.go b/pkg/vanflow/eventsource/routing.go
new file mode 100644
index 00000000..b6e78b42
--- /dev/null
+++ b/pkg/vanflow/eventsource/routing.go
@@ -0,0 +1,24 @@
+package eventsource
+
+import (
+	"github.com/skupperproject/skupper/pkg/vanflow"
+	"github.com/skupperproject/skupper/pkg/vanflow/store"
+)
+
+type RecordStoreMap map[string]store.Interface
+
+type RecordStoreRouter struct {
+	Source store.SourceRef
+	Stores RecordStoreMap
+}
+
+func (r RecordStoreRouter) Route(msg vanflow.RecordMessage) {
+	for _, record := range msg.Records {
+		typ := record.GetTypeMeta().String()
+		acc, found := r.Stores[typ]
+		if !found {
+			continue
+		}
+		acc.Patch(record, r.Source)
+	}
+}
diff --git a/pkg/vanflow/eventsource/source.go b/pkg/vanflow/eventsource/source.go
new file mode 100644
index 00000000..1451e826
--- /dev/null
+++ b/pkg/vanflow/eventsource/source.go
@@ -0,0 +1,15 @@
+package eventsource
+
+import (
+	"time"
+)
+
+// Info describes a vanflow event source
+type Info struct {
+	ID       string
+	Version  int
+	Type     string
+	Address  string
+	Direct   string
+	LastSeen time.Time
+}
diff --git a/pkg/vanflow/messages.go b/pkg/vanflow/messages.go
new file mode 100644
index 00000000..bfaa645e
--- /dev/null
+++ b/pkg/vanflow/messages.go
@@ -0,0 +1,232 @@
+package vanflow
+
+import (
+	"errors"
+	"fmt"
+
+	amqp "github.com/Azure/go-amqp"
+	"github.com/skupperproject/skupper/pkg/vanflow/encoding"
+)
+
+var (
+	beaconAddress    = "mc/sfe.all"
+	beaconSubject    = "BEACON"
+	heartbeatSubject = "HEARTBEAT"
+	flushSubject     = "FLUSH"
+	recordSubject    = "RECORD"
+)
+
+// Decode a raw amqp message into one of BeaconMessage, FlushMessage,
+// HeartbeatMessage or RecordMessage.
+func Decode(msg *amqp.Message) (interface{}, error) {
+	if msg == nil || msg.Properties == nil {
+		return nil, errors.New("cannot decode message with nil properties")
+	}
+	if msg.Properties.Subject == nil {
+		return nil, errors.New("cannot decode message without subject")
+	}
+	subject := *msg.Properties.Subject
+	switch subject {
+	case "BEACON":
+		return DecodeBeacon(msg), nil
+	case "HEARTBEAT":
+		return DecodeHeartbeat(msg), nil
+	case "FLUSH":
+		return DecodeFlush(msg), nil
+	case "RECORD":
+		return DecodeRecord(msg)
+	default:
+		return nil, fmt.Errorf("cannot decode message with subject %q", subject)
+	}
+}
+
+type BeaconMessage struct {
+	MessageProps
+	Version    uint32
+	SourceType string
+	Address    string
+	Direct     string
+	Identity   string
+}
+
+func DecodeBeacon(msg *amqp.Message) BeaconMessage {
+	var m BeaconMessage
+	if msg.Properties.To != nil {
+		m.To = *msg.Properties.To
+	}
+	if msg.Properties.Subject != nil {
+		m.Subject = *msg.Properties.Subject
+	}
+	if msg.Properties.ReplyTo != nil {
+		m.ReplyTo = *msg.Properties.ReplyTo
+	}
+	if version, ok := msg.ApplicationProperties["v"].(uint32); ok {
+		m.Version = version
+	}
+	if sourceType, ok := msg.ApplicationProperties["sourceType"].(string); ok {
+		m.SourceType = sourceType
+	}
+	if address, ok := msg.ApplicationProperties["address"].(string); ok {
+		m.Address = address
+	}
+	if direct, ok := msg.ApplicationProperties["direct"].(string); ok {
+		m.Direct = direct
+	}
+	if identity, ok := msg.ApplicationProperties["id"].(string); ok {
+		m.Identity = identity
+	}
+	return m
+}
+
+func (m BeaconMessage) Encode() *amqp.Message {
+	return &amqp.Message{
+		Properties: &amqp.MessageProperties{
+			To:      &beaconAddress,
+			Subject: &beaconSubject,
+		},
+		ApplicationProperties: map[string]interface{}{
+			"v":          m.Version,
+			"sourceType": m.SourceType,
+			"address":    m.Address,
+			"direct":     m.Direct,
+			"id":         m.Identity,
+		},
+	}
+}
+
+type MessageProps struct {
+	To      string
+	Subject string
+	ReplyTo string
+}
+
+type HeartbeatMessage struct {
+	MessageProps
+	Identity string
+	Version  uint32
+	Now      uint64
+}
+
+func DecodeHeartbeat(msg *amqp.Message) HeartbeatMessage {
+	var m HeartbeatMessage
+	if msg.Properties.To != nil {
+		m.To = *msg.Properties.To
+	}
+	if msg.Properties.Subject != nil {
+		m.Subject = *msg.Properties.Subject
+	}
+
+	if version, ok := msg.ApplicationProperties["v"].(uint32); ok {
+		m.Version = version
+	}
+	if now, ok := msg.ApplicationProperties["now"].(uint64); ok {
+		m.Now = now
+	}
+	if identity, ok := msg.ApplicationProperties["id"].(string); ok {
+		m.Identity = identity
+	}
+	return m
+}
+
+func (m HeartbeatMessage) Encode() *amqp.Message {
+	target := "mc/sfe." + m.Identity
+	return &amqp.Message{
+		Properties: &amqp.MessageProperties{
+			To:      &target,
+			Subject: &heartbeatSubject,
+		},
+		ApplicationProperties: map[string]interface{}{
+			"v":   m.Version,
+			"now": m.Now,
+			"id":  m.Identity,
+		},
+	}
+}
+
+type FlushMessage struct {
+	MessageProps
+}
+
+func DecodeFlush(msg *amqp.Message) FlushMessage {
+	var flush FlushMessage
+	if msg.Properties.To != nil {
+		flush.To = *msg.Properties.To
+	}
+	if msg.Properties.Subject != nil {
+		flush.Subject = *msg.Properties.Subject
+	}
+	if msg.Properties.ReplyTo != nil {
+		flush.ReplyTo = *msg.Properties.ReplyTo
+	}
+	return flush
+}
+
+func (m FlushMessage) Encode() *amqp.Message {
+	return &amqp.Message{
+		Properties: &amqp.MessageProperties{
+			To:      &m.To,
+			ReplyTo: &m.ReplyTo,
+			Subject: &flushSubject,
+		},
+	}
+}
+
+type RecordMessage struct {
+	MessageProps
+	Records []Record
+}
+
+func DecodeRecord(msg *amqp.Message) (record RecordMessage, err error) {
+	if msg.Properties.To != nil {
+		record.To = *msg.Properties.To
+	}
+	if msg.Properties.Subject != nil {
+		record.Subject = *msg.Properties.Subject
+	}
+	record.Records, err = decodeRecords(msg)
+	return record, err
+}
+
+func (m RecordMessage) Encode() (*amqp.Message, error) {
+	var records []interface{}
+	for i, record := range m.Records {
+		recordAttrs, err := encoding.Encode(record)
+		if err != nil {
+			return nil, fmt.Errorf("error encoding record %d: %s", i, err)
+		}
+		records = append(records, recordAttrs)
+	}
+	return &amqp.Message{
+		Properties: &amqp.MessageProperties{
+			To:      &m.To,
+			Subject: &recordSubject,
+		},
+		Value: records,
+	}, nil
+}
+
+// decodeRecords decodes an AMQP Message into a set of Records. Uses the
+// recordDecoders map to find the correct decoder for each record type.
+func decodeRecords(msg *amqp.Message) ([]Record, error) {
+	var records []Record
+	values, ok := msg.Value.([]interface{})
+	if !ok {
+		return records, fmt.Errorf("unexpected type for message Value: %T", msg.Value)
+	}
+	for _, value := range values {
+		recordAttributes, ok := value.(map[interface{}]interface{})
+		if !ok {
+			return records, fmt.Errorf("unexpected type in message Value slice: %T", value)
+		}
+		record, err := encoding.Decode(recordAttributes)
+		if err != nil {
+			return records, err
+		}
+		r, ok := record.(Record)
+		if !ok {
+			return records, fmt.Errorf("decoded type does not implement Record: %T", record)
+		}
+		records = append(records, r)
+	}
+	return records, nil
+}
diff --git a/pkg/vanflow/records.go b/pkg/vanflow/records.go
new file mode 100644
index 00000000..19aed9b1
--- /dev/null
+++ b/pkg/vanflow/records.go
@@ -0,0 +1,245 @@
+package vanflow
+
+import "github.com/skupperproject/skupper/pkg/vanflow/encoding"
+
+const apiVersion = "flow/v1"
+
+func init() {
+	encoding.MustRegisterRecord(0, SiteRecord{})
+	encoding.MustRegisterRecord(1, RouterRecord{})
+	encoding.MustRegisterRecord(2, LinkRecord{})
+	encoding.MustRegisterRecord(3, ControllerRecord{})
+	encoding.MustRegisterRecord(4, ListenerRecord{})
+	encoding.MustRegisterRecord(5, ConnectorRecord{})
+	encoding.MustRegisterRecord(6, FlowRecord{})
+	encoding.MustRegisterRecord(7, ProcessRecord{})
+	encoding.MustRegisterRecord(8, ImageRecord{})
+	encoding.MustRegisterRecord(9, IngressRecord{})
+	encoding.MustRegisterRecord(10, EgressRecord{})
+	encoding.MustRegisterRecord(11, CollectorRecord{})
+	encoding.MustRegisterRecord(12, ProcessGroupRecord{})
+	encoding.MustRegisterRecord(13, HostRecord{})
+	encoding.MustRegisterRecord(14, LogRecord{})
+	encoding.MustRegisterRecord(15, RouterAccessRecord{})
+}
+
+type SiteRecord struct {
+	BaseRecord
+	Location  *string `vflow:"9"`
+	Provider  *string `vflow:"10"`
+	Platform  *string `vflow:"11"`
+	Namespace *string `vflow:"12"`
+	Name      *string `vflow:"30"` //unspeced
+	Version   *string `vflow:"52"`
+	Policy    *string `vflow:"53"`
+}
+
+func (r SiteRecord) GetTypeMeta() TypeMeta {
+	return TypeMeta{
+		APIVersion: apiVersion,
+		Type:       "SiteRecord",
+	}
+}
+
+type RouterRecord struct {
+	BaseRecord
+	Parent       *string `vflow:"2"`
+	Namespace    *string `vflow:"12"`
+	Mode         *string `vflow:"13"`
+	ImageName    *string `vflow:"20"`
+	ImageVersion *string `vflow:"21"`
+	Hostname     *string `vflow:"22"`
+	Name         *string `vflow:"30"`
+	BuildVersion *string `vflow:"32"`
+}
+
+func (r RouterRecord) GetTypeMeta() TypeMeta {
+	return TypeMeta{
+		APIVersion: apiVersion,
+		Type:       "RouterRecord",
+	}
+}
+
+type LinkRecord struct {
+	BaseRecord
+	Parent   *string `vflow:"2"`
+	Name     *string `vflow:"30"`
+	LinkCost *uint64 `vflow:"33"`
+	// Mode      *string `vflow:"13"`
+	// Direction *string `vflow:"34"`
+	Peer   *string `vflow:"6"`
+	Role   *string `vflow:"54"`
+	Status *string `vflow:"53"`
+}
+
+func (r LinkRecord) GetTypeMeta() TypeMeta {
+	return TypeMeta{
+		APIVersion: apiVersion,
+		Type:       "LinkRecord",
+	}
+}
+
+type ControllerRecord struct {
+	BaseRecord
+	Parent       *string `vflow:"2"`
+	ImageName    *string `vflow:"20"`
+	ImageVersion *string `vflow:"21"`
+	Hostname     *string `vflow:"22"`
+	Name         *string `vflow:"30"`
+	BuildVersion *string `vflow:"32"`
+}
+
+func (r ControllerRecord) GetTypeMeta() TypeMeta {
+	return TypeMeta{
+		APIVersion: apiVersion,
+		Type:       "ControllerRecord",
+	}
+}
+
+type ListenerRecord struct {
+	BaseRecord
+	Parent      *string `vflow:"2"`
+	DestHost    *string `vflow:"15"`
+	Protocol    *string `vflow:"16"`
+	DestPort    *string `vflow:"18"`
+	Address     *string `vflow:"19"`
+	FlowCountL4 *uint64 `vflow:"40"`
+	FlowCountL7 *uint64 `vflow:"41"`
+	FlowRateL4  *uint64 `vflow:"42"`
+	FlowRateL7  *uint64 `vflow:"43"`
+}
+
+func (r ListenerRecord) GetTypeMeta() TypeMeta {
+	return TypeMeta{
+		APIVersion: apiVersion,
+		Type:       "ListenerRecord",
+	}
+}
+
+type ConnectorRecord struct {
+	BaseRecord
+	Parent      *string `vflow:"2"`
+	ProcessID   *string `vflow:"7"`
+	DestHost    *string `vflow:"15"`
+	Protocol    *string `vflow:"16"`
+	DestPort    *string `vflow:"18"`
+	Address     *string `vflow:"19"`
+	FlowCountL4 *uint64 `vflow:"40"`
+	FlowCountL7 *uint64 `vflow:"41"`
+	FlowRateL4  *uint64 `vflow:"42"`
+	FlowRateL7  *uint64 `vflow:"43"`
+}
+
+func (r ConnectorRecord) GetTypeMeta() TypeMeta {
+	return TypeMeta{
+		APIVersion: apiVersion,
+		Type:       "ConnectorRecord",
+	}
+}
+
+type FlowRecord struct {
+	BaseRecord
+	Parent         *string `vflow:"2"`
+	Counterflow    *string `vflow:"5"`
+	SourceHost     *string `vflow:"14"`
+	SourcePort     *string `vflow:"17"`
+	Octets         *uint64 `vflow:"23"`
+	Latency        *uint64 `vflow:"24"`
+	Reason         *string `vflow:"29"`
+	Trace          *string `vflow:"31"`
+	OctetRate      *uint64 `vflow:"35"`
+	OctetsOut      *uint64 `vflow:"36"`
+	OctetsUnacked  *uint64 `vflow:"37"`
+	WindowClosures *uint64 `vflow:"38"`
+	WindowSize     *uint64 `vflow:"39"`
+	Method         *string `vflow:"27"`
+	Result         *string `vflow:"28"`
+}
+
+func (r FlowRecord) GetTypeMeta() TypeMeta {
+	return TypeMeta{
+		APIVersion: apiVersion,
+		Type:       "FlowRecord",
+	}
+}
+
+type ProcessRecord struct {
+	BaseRecord
+	Parent       *string `vflow:"2"` //unspeced
+	Mode         *string `vflow:"13"`
+	SourceHost   *string `vflow:"14"`
+	ImageName    *string `vflow:"20"`
+	ImageVersion *string `vflow:"21"`
+	Hostname     *string `vflow:"22"`
+	Name         *string `vflow:"30"`
+	Group        *string `vflow:"46"`
+}
+
+func (r ProcessRecord) GetTypeMeta() TypeMeta {
+	return TypeMeta{
+		APIVersion: apiVersion,
+		Type:       "ProcessRecord",
+	}
+}
+
+type ImageRecord struct {
+	BaseRecord
+}
+
+type IngressRecord struct {
+	BaseRecord
+}
+
+type EgressRecord struct {
+	BaseRecord
+}
+
+type CollectorRecord struct {
+	BaseRecord
+}
+
+type ProcessGroupRecord struct {
+	BaseRecord
+}
+
+type HostRecord struct {
+	BaseRecord
+	Provider *string `vflow:"10"`
+	Name     *string `vflow:"30"`
+}
+
+func (r HostRecord) GetTypeMeta() TypeMeta {
+	return TypeMeta{
+		APIVersion: apiVersion,
+		Type:       "HostRecord",
+	}
+}
+
+type LogRecord struct {
+	BaseRecord
+	LogSeverity *uint64 `vflow:"48"`
+	LogText     *string `vflow:"49"`
+	SourceFile  *string `vflow:"50"`
+	SourceLine  *uint64 `vflow:"51"`
+}
+
+func (r LogRecord) GetTypeMeta() TypeMeta {
+	return TypeMeta{
+		APIVersion: apiVersion,
+		Type:       "LogRecord",
+	}
+}
+
+type RouterAccessRecord struct {
+	BaseRecord
+	Parent    *string `vflow:"2"`
+	LinkCount *uint64 `vflow:"52"`
+	Role      *string `vflow:"54"`
+}
+
+func (r RouterAccessRecord) GetTypeMeta() TypeMeta {
+	return TypeMeta{
+		APIVersion: apiVersion,
+		Type:       "RouterAccessRecord",
+	}
+}
diff --git a/pkg/vanflow/session/container.go b/pkg/vanflow/session/container.go
new file mode 100644
index 00000000..a8116c7b
--- /dev/null
+++ b/pkg/vanflow/session/container.go
@@ -0,0 +1,457 @@
+// session implements amqp connection and session management though the concept
+// of a Container, inspired by the Conatiner interface exposed by the qpid
+// proton amqp libraries. This abstraction allows vanflow components to be
+// written without repeating connection management tasks.
+package session
+
+import (
+	"context"
+	"crypto/tls"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/Azure/go-amqp"
+	"github.com/cenkalti/backoff/v4"
+)
+
+type ReceiverOptions struct {
+	Credit int
+}
+
+func (o ReceiverOptions) get() amqp.ReceiverOptions {
+	var result amqp.ReceiverOptions
+	if o.Credit <= 0 {
+		result.Credit = int32(o.Credit)
+	}
+	return result
+}
+
+type SenderOptions struct {
+}
+
+func (o SenderOptions) get() amqp.SenderOptions {
+	var result amqp.SenderOptions
+	return result
+}
+
+type Container interface {
+	// Start the container
+	Start(context.Context)
+	// NewReceiver adds a new reciver link using the container's session
+	NewReceiver(address string, opts ReceiverOptions) Receiver
+	// NewSender adds a new sender link using the container's session
+	NewSender(address string, opts SenderOptions) Sender
+	// OnSessionError
+	OnSessionError(func(err error))
+}
+
+type RetryableError interface {
+	Retry() time.Duration
+}
+
+type Receiver interface {
+	Next(context.Context) (*amqp.Message, error)
+	Accept(context.Context, *amqp.Message) error
+	Close(context.Context) error
+}
+
+type Sender interface {
+	Send(context.Context, *amqp.Message) error
+	Close(context.Context) error
+}
+
+type SASLType string
+
+const (
+	SASLTypeExternal SASLType = "EXTERNAL"
+)
+
+type ContainerConfig struct {
+	ContainerID  string
+	MaxFrameSize uint32
+	TLSConfig    *tls.Config
+	SASLType     SASLType
+	// BackOff strategy to use when reestablishing a connection defaults to an
+	// exponential backoff capped at 30 second intervals with no set retry
+	// limit.
+	BackOff backoff.BackOff
+}
+
+func (cfg ContainerConfig) toAmqp() *amqp.ConnOptions {
+	opts := amqp.ConnOptions{
+		ContainerID:  cfg.ContainerID,
+		MaxFrameSize: cfg.MaxFrameSize,
+		TLSConfig:    cfg.TLSConfig,
+	}
+	switch cfg.SASLType {
+	case SASLTypeExternal:
+		opts.SASLType = amqp.SASLTypeExternal("")
+	}
+	return &opts
+}
+
+// NewContainer creats an amqp container that will attempt to create a single
+// connection + session pair using the supplied amqp connection options for use
+// with the container's Senders and Receivers. Will recreate the connection and
+// session when a link encounters an error using the specified backoff
+// strategy.
+func NewContainer(address string, config ContainerConfig) Container {
+	c := &container{
+		address:       address,
+		config:        config,
+		hasNext:       make(chan struct{}),
+		sessionErrors: make(chan sessionErr, 32),
+		notifyOK:      make(chan int, 32),
+	}
+	return c
+}
+
+type container struct {
+	address string
+	config  ContainerConfig
+
+	mu            sync.Mutex
+	sess          *amqp.Session
+	gen           int
+	hasNext       chan struct{}
+	errorHandlers []func(error)
+
+	sessionErrors chan sessionErr
+	notifyOK      chan int
+}
+
+func (c *container) OnSessionError(handler func(error)) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.errorHandlers = append(c.errorHandlers, handler)
+}
+
+func (c *container) awaitNextSession(ctx context.Context, prev int) (session *amqp.Session, gen int, err error) {
+	c.mu.Lock()
+	session, gen, hasNext := c.sess, c.gen, c.hasNext
+	c.mu.Unlock()
+	if prev != gen {
+		return session, gen, nil
+	}
+	select {
+	case <-ctx.Done():
+		return nil, 0, ctx.Err()
+	case <-hasNext:
+		return c.awaitNextSession(ctx, prev)
+	}
+}
+
+type sessionErr struct {
+	Generation int
+	Err        error
+}
+
+func (e sessionErr) Error() string {
+	return e.Err.Error()
+}
+
+// Start the container. It will run until the context is cancelled or until the
+// backoff strategy chosen finishes.
+func (c *container) Start(ctx context.Context) {
+	if c.config.BackOff == nil {
+		b := backoff.NewExponentialBackOff()
+		b.InitialInterval = time.Millisecond * 250
+		b.MaxInterval = time.Second * 30
+		b.MaxElapsedTime = 0
+		b.Reset()
+		c.config.BackOff = b
+	}
+
+	go func() {
+		var generation int
+		var prevSessionTeardown func() = func() {}
+		b := backoff.WithContext(c.config.BackOff, ctx)
+		err := backoff.RetryNotify(
+			func() error {
+				conn, err := amqp.Dial(ctx, c.address, c.config.toAmqp())
+				if err != nil {
+					return fmt.Errorf("dial error: %s", err)
+				}
+				sess, err := conn.NewSession(ctx, nil)
+				if err != nil {
+					return fmt.Errorf("session create error: %s", err)
+				}
+				generation++
+
+				c.mu.Lock()
+				close(c.hasNext)
+				c.sess, c.gen, c.hasNext = sess, generation, make(chan struct{})
+				c.mu.Unlock()
+
+				prevSessionTeardown()
+				prevSessionTeardown = func() {
+					sess.Close(ctx)
+					conn.Close()
+				}
+
+				for {
+					select {
+					case <-ctx.Done():
+						return ctx.Err()
+					case gen := <-c.notifyOK:
+						if gen == generation {
+							b.Reset()
+						}
+					case recvErr := <-c.sessionErrors:
+						if recvErr.Generation == generation {
+							return fmt.Errorf("session receiver error: %s", recvErr)
+						}
+					}
+				}
+			},
+			b,
+			func(err error, d time.Duration) {
+				wErr := errSessionRestart{Err: err, D: d}
+				c.mu.Lock()
+				for _, handler := range c.errorHandlers {
+					handler(wErr)
+				}
+				c.mu.Unlock()
+			},
+		)
+		defer prevSessionTeardown()
+		if err != nil {
+			if errors.Is(err, ctx.Err()) {
+				return
+			}
+			wErr := fmt.Errorf("error caused contianer to close: %w", err)
+			c.mu.Lock()
+			defer c.mu.Unlock()
+			for _, handler := range c.errorHandlers {
+				handler(wErr)
+			}
+		}
+	}()
+}
+
+type errSessionRestart struct {
+	Err error
+	D   time.Duration
+}
+
+func (e errSessionRestart) Error() string {
+	return fmt.Sprintf("session error: %s", e.Err)
+}
+
+func (e errSessionRestart) Retry() time.Duration {
+	return e.D
+}
+
+func (s *container) NewReceiver(address string, opts ReceiverOptions) Receiver {
+	return s.newLink(address, opts, SenderOptions{})
+}
+
+func (s *container) NewSender(address string, opts SenderOptions) Sender {
+	return s.newLink(address, ReceiverOptions{}, opts)
+}
+
+func (c *container) newLink(address string, r ReceiverOptions, s SenderOptions) *link {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	l := &link{
+		address:       address,
+		container:     c,
+		receiverOpts:  r.get(),
+		senderOpts:    s.get(),
+		sessionErrors: c.sessionErrors,
+		reportOK:      c.notifyOK,
+		curr:          c.sess,
+		currGen:       c.gen,
+	}
+	return l
+}
+
+type link struct {
+	address      string
+	receiverOpts amqp.ReceiverOptions
+	senderOpts   amqp.SenderOptions
+
+	sessionErrors chan<- sessionErr
+	reportOK      chan<- int
+
+	container *container
+
+	mu      sync.Mutex
+	closed  bool
+	currGen int
+	curr    *amqp.Session
+	rcvGen  int
+	rcv     *amqp.Receiver
+	sndGen  int
+	snd     *amqp.Sender
+}
+
+var errLinkClosed = errors.New("link closed")
+
+func (r *link) awaitSession(ctx context.Context) error {
+	r.mu.Lock()
+	curr, currGen, closed := r.curr, r.currGen, r.closed
+	r.mu.Unlock()
+	if closed {
+		return errLinkClosed
+	}
+	if curr != nil {
+		return nil
+	}
+	next, nextGen, err := r.container.awaitNextSession(ctx, currGen)
+	if err != nil {
+		return err
+	}
+	r.mu.Lock()
+	r.curr, r.currGen = next, nextGen
+	r.mu.Unlock()
+	return nil
+}
+
+func (r *link) withReceiver(ctx context.Context, fn func(receiver *amqp.Receiver, generation int) error) error {
+	if err := r.awaitSession(ctx); err != nil {
+		return r.handleError(ctx, fmt.Errorf("session await error: %w", err))
+	}
+	rcv, currGen, err := r.getReceiver(ctx)
+	if err != nil {
+		return r.handleError(ctx, fmt.Errorf("receiver create error: %w", err))
+	}
+	if err := fn(rcv, currGen); err != nil {
+		return r.handleError(ctx, err)
+	}
+	return nil
+}
+
+func (r *link) withSender(ctx context.Context, fn func(sender *amqp.Sender, generation int) error) error {
+	if err := r.awaitSession(ctx); err != nil {
+		return r.handleError(ctx, fmt.Errorf("session await error: %w", err))
+	}
+	snd, currGen, err := r.getSender(ctx)
+	if err != nil {
+		return r.handleError(ctx, fmt.Errorf("sender create error: %w", err))
+	}
+	if err := fn(snd, currGen); err != nil {
+		return r.handleError(ctx, err)
+	}
+	return nil
+}
+
+func (r *link) handleError(ctx context.Context, err error) error {
+	if errors.Is(err, ctx.Err()) {
+		return err
+	}
+	if errors.Is(err, errLinkClosed) {
+		return err
+	}
+	r.sessionErrors <- sessionErr{
+		Generation: r.currGen,
+		Err:        err,
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.curr, r.rcv, r.snd = nil, nil, nil
+	return err
+}
+
+func (r *link) getReceiver(ctx context.Context) (*amqp.Receiver, int, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.rcv != nil && r.rcvGen == r.currGen {
+		return r.rcv, r.rcvGen, nil
+	}
+	rcv, err := r.curr.NewReceiver(ctx, r.address, &amqp.ReceiverOptions{Credit: int32(r.receiverOpts.Credit)})
+	if err != nil {
+		return nil, 0, err
+	}
+	r.rcv, r.rcvGen = rcv, r.currGen
+	return r.rcv, r.rcvGen, nil
+}
+
+func (r *link) getSender(ctx context.Context) (*amqp.Sender, int, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.snd != nil && r.sndGen == r.currGen {
+		return r.snd, r.sndGen, nil
+	}
+	snd, err := r.curr.NewSender(ctx, r.address, &r.senderOpts)
+	if err != nil {
+		return nil, 0, err
+	}
+	r.snd, r.sndGen = snd, r.currGen
+	return r.snd, r.sndGen, nil
+}
+
+func (r *link) Next(ctx context.Context) (*amqp.Message, error) {
+	var (
+		result *amqp.Message
+	)
+	err := r.withReceiver(ctx, func(rcv *amqp.Receiver, _ int) error {
+		msg, err := rcv.Receive(ctx, nil)
+		if err != nil {
+			return fmt.Errorf("receive error: %w", err)
+		}
+		result = msg
+		return nil
+	})
+	if err != nil && ctx.Err() == nil {
+		return r.Next(ctx)
+	}
+	return result, err
+}
+
+func (r *link) Accept(ctx context.Context, msg *amqp.Message) error {
+	var (
+		acceptErr  error
+		generation int
+	)
+	err := r.withReceiver(ctx, func(rcv *amqp.Receiver, gen int) error {
+		generation = gen
+		acceptErr = rcv.AcceptMessage(ctx, msg)
+		// accepting messages is a stateful operation, so don't report the error back
+		// to the container and signal a connection teardown
+		return nil
+	})
+	if acceptErr != nil {
+		return acceptErr
+	}
+	if err == nil {
+		select {
+		case r.reportOK <- generation:
+		default:
+		}
+	}
+	return err
+}
+
+func (r *link) Send(ctx context.Context, msg *amqp.Message) error {
+	var generation int
+	err := r.withSender(ctx, func(snd *amqp.Sender, gen int) error {
+		generation = gen
+		err := snd.Send(ctx, msg, nil)
+		if err != nil {
+			return fmt.Errorf("send error: %w", err)
+		}
+		return nil
+	})
+	if err == nil {
+		select {
+		case r.reportOK <- generation:
+		default:
+		}
+	}
+	return err
+}
+
+func (r *link) Close(ctx context.Context) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.closed = true
+	rcv := r.rcv
+	r.curr, r.rcv, r.snd = nil, nil, nil
+	if rcv != nil {
+		return rcv.Close(ctx)
+	}
+	return nil
+}
diff --git a/pkg/vanflow/session/factory.go b/pkg/vanflow/session/factory.go
new file mode 100644
index 00000000..647fb4c9
--- /dev/null
+++ b/pkg/vanflow/session/factory.go
@@ -0,0 +1,46 @@
+package session
+
+import (
+	"crypto/rand"
+	"encoding/hex"
+	"io"
+)
+
+// ContainerFactory creates Containers with a given configuration
+type ContainerFactory interface {
+	Create() Container
+}
+
+func NewContainerFactory(address string, cfg ContainerConfig) ContainerFactory {
+	return factory{
+		Address: address,
+		Config:  cfg,
+	}
+}
+
+func NewMockContainerFactory() ContainerFactory {
+	return mockFactory{Router: NewMockRouter()}
+}
+
+type mockFactory struct {
+	Router *MockRouter
+}
+
+func (m mockFactory) Create() Container {
+	return NewMockContainer(m.Router)
+}
+
+type factory struct {
+	Address string
+	Config  ContainerConfig
+}
+
+func (f factory) Create() Container {
+	return NewContainer(f.Address, f.Config)
+}
+
+func randomID() string {
+	var bytes [16]byte
+	io.ReadFull(rand.Reader, bytes[:])
+	return hex.EncodeToString(bytes[:])
+}
diff --git a/pkg/vanflow/session/mock.go b/pkg/vanflow/session/mock.go
new file mode 100644
index 00000000..fced8ad0
--- /dev/null
+++ b/pkg/vanflow/session/mock.go
@@ -0,0 +1,175 @@
+package session
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"github.com/Azure/go-amqp"
+)
+
+// NewMockContainer creates a mock container using the mock router
+func NewMockContainer(router *MockRouter) Container {
+	return &mockContainer{
+		Router: router,
+	}
+}
+
+type mockContainer struct {
+	Router *MockRouter
+}
+
+func (c *mockContainer) Start(ctx context.Context) {
+}
+
+func (c *mockContainer) OnSessionError(func(error)) {
+}
+
+func (c *mockContainer) NewReceiver(address string, opts ReceiverOptions) Receiver {
+	cred := opts.Credit
+	if cred <= 0 {
+		cred = 256
+	}
+	rcv := &mockReceiver{
+		channel: make(chan *amqp.Message, cred),
+		done:    make(chan struct{}),
+	}
+	c.Router.subscribe(address, rcv)
+	return rcv
+}
+
+func (c *mockContainer) NewSender(address string, opts SenderOptions) Sender {
+	return &mockSender{
+		router:  c.Router,
+		address: address,
+		done:    make(chan struct{}),
+	}
+}
+
+// NewMockRouter returns a mock router that can be used by containers to send
+// and receive multicast messages. Behaves like a very rough approximation of
+// the skupper router for routing vanflow messages.
+func NewMockRouter() *MockRouter {
+	return &MockRouter{
+		topics: make(map[string]*multicast),
+	}
+}
+
+type MockRouter struct {
+	mu     sync.Mutex
+	topics map[string]*multicast
+}
+
+func (router *MockRouter) subscribe(address string, r *mockReceiver) {
+	router.get(address).subscribe(r)
+}
+
+func (router *MockRouter) send(ctx context.Context, address string, msg *amqp.Message) error {
+	return router.get(address).send(ctx, msg)
+}
+
+func (router *MockRouter) get(address string) *multicast {
+	router.mu.Lock()
+	defer router.mu.Unlock()
+	t, ok := router.topics[address]
+	if !ok {
+		t = newMulticast()
+		router.topics[address] = t
+	}
+	return t
+}
+
+type multicast struct {
+	mu        sync.Mutex
+	receivers []*mockReceiver
+	update    chan struct{}
+}
+
+func newMulticast() *multicast {
+	return &multicast{update: make(chan struct{})}
+}
+
+func (m *multicast) send(ctx context.Context, msg *amqp.Message) error {
+	m.mu.Lock()
+	update := m.update
+	var sent bool
+	for _, r := range m.receivers {
+		sent = true
+		r.send(msg)
+	}
+	m.mu.Unlock()
+	if !sent {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-update:
+			return m.send(ctx, msg)
+		}
+	}
+	return nil
+}
+
+func (m *multicast) subscribe(r *mockReceiver) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.receivers = append(m.receivers, r)
+	close(m.update)
+	m.update = make(chan struct{})
+}
+
+type mockReceiver struct {
+	channel   chan *amqp.Message
+	closeOnce sync.Once
+	done      chan struct{}
+}
+
+func (r *mockReceiver) send(msg *amqp.Message) {
+	select {
+	case <-r.done:
+		return
+	default:
+		r.channel <- msg
+	}
+}
+func (r *mockReceiver) Next(ctx context.Context) (*amqp.Message, error) {
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case <-r.done:
+		return nil, fmt.Errorf("closed")
+	case msg := <-r.channel:
+		return msg, nil
+	}
+}
+
+func (r *mockReceiver) Accept(context.Context, *amqp.Message) error {
+	return nil
+}
+
+func (r *mockReceiver) Close(context.Context) error {
+	r.closeOnce.Do(func() { close(r.done) })
+	return nil
+}
+
+type mockSender struct {
+	router    *MockRouter
+	address   string
+	closeOnce sync.Once
+	done      chan struct{}
+}
+
+func (s *mockSender) Send(ctx context.Context, msg *amqp.Message) error {
+	select {
+	case <-s.done:
+		return fmt.Errorf("sender closed")
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+		return s.router.send(ctx, s.address, msg)
+	}
+}
+
+func (s *mockSender) Close(context.Context) error {
+	s.closeOnce.Do(func() { close(s.done) })
+	return nil
+}
diff --git a/pkg/vanflow/store/store.go b/pkg/vanflow/store/store.go
new file mode 100644
index 00000000..b5cfec80
--- /dev/null
+++ b/pkg/vanflow/store/store.go
@@ -0,0 +1,44 @@
+package store
+
+import (
+	"time"
+
+	"github.com/skupperproject/skupper/pkg/vanflow"
+)
+
+// Entry is the store record container
+type Entry struct {
+	Metadata
+	Record vanflow.Record
+}
+
+// Metadata about a record
+type Metadata struct {
+	LastUpdate time.Time
+
+	Source SourceRef
+}
+
+// SourceRef identifies a record source
+type SourceRef struct {
+	ID      string
+	Version string
+}
+
+// Interface to a vanflow record store
+type Interface interface {
+	Add(record vanflow.Record, source SourceRef) bool
+	Update(vanflow.Record) bool
+	Get(id string) (Entry, bool)
+	Delete(id string) (Entry, bool)
+
+	// Patch will either create a new record or will merge the partial state
+	// contained in the record with its stored state
+	Patch(record vanflow.Record, source SourceRef)
+
+	List() []Entry
+	Index(index string, exemplar Entry) []Entry
+	IndexValues(index string) []string
+
+	Replace([]Entry)
+}
diff --git a/pkg/vanflow/store/syncmap.go b/pkg/vanflow/store/syncmap.go
new file mode 100644
index 00000000..9e78f7c5
--- /dev/null
+++ b/pkg/vanflow/store/syncmap.go
@@ -0,0 +1,323 @@
+package store
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/skupperproject/skupper/pkg/vanflow"
+	"github.com/skupperproject/skupper/pkg/vanflow/encoding"
+)
+
+type EventHandlerFuncs struct {
+	OnAdd    func(entry Entry)
+	OnChange func(prev, curr Entry)
+	OnDelete func(entry Entry)
+}
+
+type syncMapStore struct {
+	mu    sync.RWMutex
+	items map[string]Entry
+
+	indexers      map[string]Indexer
+	indicies      map[string]map[string]keySet
+	eventHandlers EventHandlerFuncs
+}
+
+type SyncMapStoreConfig struct {
+	Indexers map[string]Indexer
+	Handlers EventHandlerFuncs
+}
+
+func NewSyncMapStore(cfg SyncMapStoreConfig) Interface {
+	if cfg.Indexers == nil {
+		cfg.Indexers = defaultIndexers()
+	}
+	return &syncMapStore{
+		indexers:      cfg.Indexers,
+		eventHandlers: cfg.Handlers,
+
+		items:    make(map[string]Entry),
+		indicies: make(map[string]map[string]keySet),
+	}
+}
+
+func (m *syncMapStore) Add(record vanflow.Record, source SourceRef) bool {
+
+	entry, ok := func() (Entry, bool) {
+		key := record.Identity()
+
+		entry := Entry{
+			Metadata: Metadata{LastUpdate: time.Now(), Source: source},
+			Record:   record,
+		}
+
+		m.mu.Lock()
+		defer m.mu.Unlock()
+		if _, exists := m.items[key]; exists {
+			return entry, false
+		}
+		m.items[key] = entry
+		m.reindex(key, nil, entry)
+		return entry, true
+	}()
+
+	if ok && m.eventHandlers.OnAdd != nil {
+		m.eventHandlers.OnAdd(entry)
+	}
+	return ok
+}
+
+func (m *syncMapStore) Update(record vanflow.Record) bool {
+	prev, next, ok := func() (Entry, Entry, bool) {
+		var prev, next Entry
+		key := record.Identity()
+
+		m.mu.Lock()
+		defer m.mu.Unlock()
+		prev, exists := m.items[key]
+		if !exists {
+			return prev, next, false
+		}
+		next = prev
+		next.LastUpdate = time.Now()
+		next.Record = record
+		m.items[key] = next
+		m.reindex(key, &prev, next)
+		return prev, next, true
+	}()
+
+	if ok && m.eventHandlers.OnChange != nil {
+		m.eventHandlers.OnChange(prev, next)
+	}
+
+	return ok
+}
+
+func (m *syncMapStore) Get(id string) (Entry, bool) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	entry, ok := m.items[id]
+	return entry, ok
+}
+
+func (m *syncMapStore) Delete(id string) (Entry, bool) {
+	prev, ok := func() (Entry, bool) {
+		m.mu.Lock()
+		defer m.mu.Unlock()
+		curr, exists := m.items[id]
+		if !exists {
+			return curr, false
+		}
+		delete(m.items, id)
+		m.unindex(id, curr)
+		return curr, true
+	}()
+	if ok && m.eventHandlers.OnDelete != nil {
+		m.eventHandlers.OnDelete(prev)
+	}
+	return prev, ok
+}
+
+func (m *syncMapStore) Patch(record vanflow.Record, source SourceRef) {
+
+	key := record.Identity()
+	const (
+		ok       = 0
+		dne      = 1
+		noChange = 2
+	)
+	prev, next, status, err := func() (prev Entry, next Entry, status int, err error) {
+		m.mu.Lock()
+		defer m.mu.Unlock()
+		curr, exists := m.items[key]
+		if !exists {
+			return prev, next, dne, nil
+		}
+
+		currAttrs, err := encoding.Encode(curr.Record)
+		if err != nil {
+			err = fmt.Errorf("error encoding current record for comparison: %w", err)
+			return
+		}
+		nextAttrs, err := encoding.Encode(record)
+		if err != nil {
+			err = fmt.Errorf("error encoding incoming record for comparison: %w", err)
+			return
+		}
+		var changed bool
+		for nK, nV := range nextAttrs {
+			cV, ok := currAttrs[nK]
+			if !ok || cV != nV {
+				changed = true
+				currAttrs[nK] = nV
+			}
+		}
+		if !changed {
+			return prev, next, noChange, nil
+		}
+
+		prev = curr
+		next = curr
+		patched, err := encoding.Decode(currAttrs)
+		next.Record = patched.(vanflow.Record)
+		next.LastUpdate = time.Now()
+		m.items[key] = next
+		m.reindex(key, &prev, next)
+		return
+	}()
+
+	if err != nil {
+		panic(err) // TODO Decide on error handing - shouldn't happen
+	}
+
+	switch status {
+	case dne:
+		m.Add(record, source)
+		return
+	case noChange:
+		return
+	}
+
+	if m.eventHandlers.OnChange != nil {
+		m.eventHandlers.OnChange(prev, next)
+	}
+}
+
+func (m *syncMapStore) List() []Entry {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	entries := make([]Entry, 0, len(m.items))
+	for _, entry := range m.items {
+		entries = append(entries, entry)
+	}
+	return entries
+}
+
+func (m *syncMapStore) Index(index string, exemplar Entry) []Entry {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	indexer, ok := m.indexers[index]
+	if !ok {
+		return nil
+	}
+	idx := m.indicies[index]
+	indexVals := indexer(exemplar)
+
+	keys := make(keySet)
+	for _, indexVal := range indexVals {
+		for key := range idx[indexVal] {
+			keys.Add(key)
+		}
+	}
+	entries := make([]Entry, 0, len(keys))
+	for key := range keys {
+		entries = append(entries, m.items[key])
+	}
+	return entries
+}
+func (m *syncMapStore) IndexValues(index string) []string {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	idx := m.indicies[index]
+	if len(idx) == 0 {
+		return nil
+	}
+	values := make([]string, 0, len(idx))
+	for val := range idx {
+		values = append(values, val)
+	}
+	return values
+}
+
+func (m *syncMapStore) Replace(items []Entry) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	entries := make(map[string]Entry, len(items))
+	for _, item := range items {
+		entries[item.Record.Identity()] = item
+	}
+	m.items = entries
+
+	m.indicies = make(map[string]map[string]keySet)
+	for key, entry := range m.items {
+		m.reindex(key, nil, entry)
+	}
+}
+
+func (m *syncMapStore) unindex(key string, entry Entry) {
+	for name, indexer := range m.indexers {
+		indexVals := indexer(entry)
+
+		index := m.indicies[name]
+		if index == nil {
+			continue
+		}
+		for _, val := range indexVals {
+			if set := index[val]; set != nil {
+				set.Remove(key)
+				if len(set) == 0 {
+					delete(index, val)
+				}
+			}
+		}
+	}
+}
+
+func (m *syncMapStore) reindex(key string, prev *Entry, next Entry) {
+	if prev != nil {
+		m.unindex(key, *prev)
+	}
+	for name, indexer := range m.indexers {
+		indexVals := indexer(next)
+
+		index := m.indicies[name]
+		if index == nil {
+			index = map[string]keySet{}
+			m.indicies[name] = index
+		}
+		for _, indexVal := range indexVals {
+			set := index[indexVal]
+			if set == nil {
+				set = keySet{}
+				index[indexVal] = set
+			}
+			set.Add(key)
+		}
+	}
+}
+
+type Indexer func(Entry) []string
+type keySet map[string]struct{}
+
+func (s keySet) Add(id string) {
+	s[id] = struct{}{}
+}
+func (s keySet) Remove(id string) bool {
+	_, present := s[id]
+	if present {
+		delete(s, id)
+	}
+	return present
+}
+
+const (
+	SourceIndex = "BySource"
+	TypeIndex   = "ByType"
+)
+
+func SourceIndexer(e Entry) []string {
+	return []string{fmt.Sprintf("%s/%s", e.Source.Version, e.Metadata.Source.ID)}
+}
+func TypeIndexer(e Entry) []string {
+	return []string{e.Record.GetTypeMeta().String()}
+}
+
+func defaultIndexers() map[string]Indexer {
+	return map[string]Indexer{
+		SourceIndex: SourceIndexer,
+		TypeIndex:   TypeIndexer,
+	}
+}


Test patch:
diff --git a/pkg/kube/flow/status_test.go b/pkg/kube/flow/status_test.go
new file mode 100644
index 00000000..427b0efd
--- /dev/null
+++ b/pkg/kube/flow/status_test.go
@@ -0,0 +1,187 @@
+package flow
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/skupperproject/skupper/pkg/network"
+	"github.com/skupperproject/skupper/pkg/vanflow"
+	"github.com/skupperproject/skupper/pkg/vanflow/eventsource"
+	"github.com/skupperproject/skupper/pkg/vanflow/session"
+	"github.com/skupperproject/skupper/pkg/vanflow/store"
+	"golang.org/x/time/rate"
+	"gotest.tools/poll"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+func TestStatusSyncEndToEnd(t *testing.T) {
+	ctx := context.Background()
+	cm := fake.NewSimpleClientset(&v1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test",
+			Namespace: "test",
+		},
+	}).CoreV1().ConfigMaps("test")
+
+	factory := session.NewMockContainerFactory()
+	conn := factory.Create()
+	go conn.Start(ctx)
+
+	storA := store.NewSyncMapStore(store.SyncMapStoreConfig{})
+	sourceA := eventsource.NewManager(conn, eventsource.ManagerConfig{
+		Source: eventsource.Info{
+			ID: "A", Type: "TESTING",
+			Address: "mc/sfe.A", Direct: "sfe.A",
+		},
+		HeartbeatInterval: 5 * time.Millisecond, BeaconInterval: 50 * time.Millisecond,
+		Stores: []store.Interface{storA},
+	})
+	storB := store.NewSyncMapStore(store.SyncMapStoreConfig{})
+	sourceB := eventsource.NewManager(conn, eventsource.ManagerConfig{
+		Source: eventsource.Info{
+			ID: "B", Type: "TESTING",
+			Address: "mc/sfe.B", Direct: "sfe.B",
+		},
+		HeartbeatInterval: 5 * time.Millisecond, BeaconInterval: 50 * time.Millisecond,
+		Stores: []store.Interface{storB},
+	})
+
+	go sourceA.Run(ctx)
+	go sourceB.Run(ctx)
+
+	testCases := []struct {
+		Rate     time.Duration
+		RecordsA []vanflow.Record
+		RecordsB []vanflow.Record
+		Expected network.NetworkStatusInfo
+	}{
+		{
+			RecordsA: []vanflow.Record{
+				vanflow.SiteRecord{BaseRecord: vanflow.NewBase("site-a")},
+			},
+			Expected: network.NetworkStatusInfo{SiteStatus: []network.SiteStatusInfo{
+				{Site: network.SiteInfo{Identity: "site-a"}},
+			}},
+		},
+		{
+			Rate: time.Millisecond,
+			RecordsA: []vanflow.Record{
+				vanflow.SiteRecord{
+					BaseRecord: vanflow.NewBase("site-a"),
+				},
+				vanflow.ProcessRecord{
+					BaseRecord: vanflow.NewBase("p01"), Parent: ptrTo("site-a"), Name: ptrTo("web-01"),
+				},
+				vanflow.ProcessRecord{BaseRecord: vanflow.NewBase("p02"), Parent: ptrTo("site-a"),
+					SourceHost: ptrTo("10.0.0.101"), Name: ptrTo("web-02"),
+				},
+			},
+			RecordsB: []vanflow.Record{
+				vanflow.RouterRecord{BaseRecord: vanflow.NewBase("router-a"), Parent: ptrTo("site-a"), Name: ptrTo("RouterA")},
+				vanflow.RouterRecord{BaseRecord: vanflow.NewBase("router-b"), Parent: ptrTo("site-a"), Name: ptrTo("RouterB")},
+				vanflow.ConnectorRecord{
+					BaseRecord: vanflow.NewBase("connector01"), Parent: ptrTo("router-a"),
+					ProcessID: ptrTo("p01"), Address: ptrTo("web"), DestHost: ptrTo("10.0.0.100"),
+				},
+				vanflow.ConnectorRecord{
+					BaseRecord: vanflow.NewBase("connector02"), Parent: ptrTo("router-a"),
+					Address: ptrTo("web"), DestHost: ptrTo("10.0.0.101"),
+				},
+				vanflow.ListenerRecord{
+					BaseRecord: vanflow.NewBase("listener01"), Parent: ptrTo("router-a"),
+					Address: ptrTo("web"),
+				},
+				vanflow.LinkRecord{
+					BaseRecord: vanflow.NewBase("automeshlink-BtoA"), Parent: ptrTo("router-b"),
+					Peer: ptrTo("routeraccess-a"), Name: ptrTo("auto-mesh-connector/0"),
+				},
+			},
+			Expected: network.NetworkStatusInfo{
+				Addresses: []network.AddressInfo{
+					{Name: "web", ListenerCount: 1, ConnectorCount: 2},
+				},
+				SiteStatus: []network.SiteStatusInfo{
+					{
+						Site: network.SiteInfo{Identity: "site-a"},
+						RouterStatus: []network.RouterStatusInfo{
+							{
+								Router: network.RouterInfo{Name: "RouterA"},
+								Links:  []network.LinkInfo{},
+								Listeners: []network.ListenerInfo{
+									{Address: "web"},
+								},
+								Connectors: []network.ConnectorInfo{
+									{Process: "p01", Target: "web-01", Address: "web", DestHost: "10.0.0.100"},
+									{Process: "p02", Target: "web-02", Address: "web", DestHost: "10.0.0.101"},
+								},
+							},
+							{Router: network.RouterInfo{Name: "RouterB"},
+								Links: []network.LinkInfo{{Name: "auto-mesh-connector/0"}},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run("", func(t *testing.T) {
+			ss := NewStatusSync(factory, nil, cm, "test")
+			if tc.Rate > 0 {
+				ss.limit = rate.Every(tc.Rate)
+			} else {
+				ss.limit = 0
+			}
+			storA.Replace(asEntries(tc.RecordsA))
+			storB.Replace(asEntries(tc.RecordsB))
+
+			tstCtx, cancel := context.WithCancel(ctx)
+			defer cancel()
+			go ss.Run(tstCtx)
+
+			poll.WaitOn(t, func(log poll.LogT) poll.Result {
+				actualCM, err := cm.Get(tstCtx, "test", metav1.GetOptions{})
+				if err != nil {
+					return poll.Error(fmt.Errorf("unexpected error condition: %s", err))
+				}
+
+				statusString, ok := actualCM.Data["NetworkStatus"]
+				if !ok {
+					return poll.Continue("NetworkStatus not set in configmap")
+				}
+
+				var actual network.NetworkStatusInfo
+				if err := json.Unmarshal([]byte(statusString), &actual); err != nil {
+					return poll.Error(fmt.Errorf("invalid json found in configmap: %s", err))
+				}
+
+				if !cmp.Equal(actual, tc.Expected, cmpopts.EquateEmpty()) {
+					return poll.Continue("actual does not match expected: %s", cmp.Diff(actual, tc.Expected, cmpopts.EquateEmpty()))
+				}
+				return poll.Success()
+			}, poll.WithDelay(time.Millisecond*10), poll.WithTimeout(time.Second*2))
+		})
+	}
+}
+
+func asEntries(r []vanflow.Record) []store.Entry {
+	entries := make([]store.Entry, len(r))
+	for i := range r {
+		entries[i] = store.Entry{
+			Record: r[i],
+		}
+	}
+	return entries
+}
+
+func ptrTo[T any, R *T](in T) R {
+	return &in
+}
diff --git a/pkg/vanflow/base_test.go b/pkg/vanflow/base_test.go
new file mode 100644
index 00000000..96ad3fa1
--- /dev/null
+++ b/pkg/vanflow/base_test.go
@@ -0,0 +1,97 @@
+package vanflow
+
+import (
+	"testing"
+	"time"
+
+	"github.com/skupperproject/skupper/pkg/vanflow/encoding"
+	"gotest.tools/assert"
+)
+
+func TestTimeEncoding(t *testing.T) {
+
+	now := time.Now().Truncate(time.Microsecond)
+	testCases := []struct {
+		Name           string
+		In             any
+		DecodeError    string
+		EncodeError    string
+		NotDeeplyEqual bool
+		HandleOutput   func(*testing.T, map[interface{}]interface{})
+	}{
+		{
+			Name: "basic",
+			In: LinkRecord{
+				BaseRecord: NewBase("i", now),
+			},
+			HandleOutput: func(t *testing.T, attrs map[interface{}]interface{}) {
+				assert.Equal(t, attrs[uint32(3)], uint64(now.UnixMicro()))
+			},
+		}, {
+			Name: "before epoch",
+			In: LinkRecord{
+				BaseRecord: NewBase("i", time.UnixMicro(-1)),
+			},
+			DecodeError: `cannot represent times before epoch in this encoding`,
+		}, {
+			Name: "too large",
+			In: LinkRecord{
+				BaseRecord: BaseRecord{ID: "i"},
+			},
+			HandleOutput: func(t *testing.T, attrs map[interface{}]interface{}) {
+				_, ok := attrs[uint32(3)]
+				assert.Assert(t, !ok)
+				attrs[uint32(3)] = uint64(1 << 63) // too large for int64
+			},
+			EncodeError: `time too far in future for internal representation`,
+		}, {
+			Name: "unexpected type",
+			In: LinkRecord{
+				BaseRecord: BaseRecord{ID: "i"},
+			},
+			HandleOutput: func(t *testing.T, attrs map[interface{}]interface{}) {
+				_, ok := attrs[uint32(3)]
+				assert.Assert(t, !ok)
+				attrs[uint32(3)] = "yahoo"
+			},
+			EncodeError: `expected type uint64 for timestamp but got string`,
+		}, {
+			Name:           "Ignores Zero Time",
+			NotDeeplyEqual: true,
+			In: LinkRecord{
+				BaseRecord: NewBase("i", time.Time{}, time.UnixMicro(0)),
+			},
+			HandleOutput: func(t *testing.T, attrs map[interface{}]interface{}) {
+				_, ok := attrs[uint32(3)]
+				assert.Assert(t, !ok)
+				endTime, ok := attrs[uint32(4)]
+				assert.Assert(t, ok)
+				assert.Equal(t, endTime, uint64(0))
+			},
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.Name, func(t *testing.T) {
+			out, err := encoding.Encode(tc.In)
+			if tc.DecodeError != "" {
+				assert.ErrorContains(t, err, tc.DecodeError)
+				return
+			}
+			assert.Check(t, err)
+			if tc.HandleOutput != nil {
+				tc.HandleOutput(t, out)
+			}
+			duplicate, err := encoding.Decode(out)
+			if tc.EncodeError != "" {
+				assert.ErrorContains(t, err, tc.EncodeError)
+				return
+			}
+			assert.Check(t, err)
+			if tc.NotDeeplyEqual {
+				return
+			}
+			assert.DeepEqual(t, tc.In, duplicate)
+		})
+	}
+
+}
diff --git a/pkg/vanflow/encoding/decoder_test.go b/pkg/vanflow/encoding/decoder_test.go
new file mode 100644
index 00000000..39ade9c1
--- /dev/null
+++ b/pkg/vanflow/encoding/decoder_test.go
@@ -0,0 +1,247 @@
+package encoding_test
+
+import (
+	"fmt"
+	"math"
+	"path"
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/skupperproject/skupper/pkg/vanflow"
+	"github.com/skupperproject/skupper/pkg/vanflow/encoding"
+)
+
+type tNestedEmbedded struct {
+	A
+	Root uint32 `vflow:"1"`
+}
+type A struct {
+	B
+	X
+}
+type B struct {
+	B string `vflow:"100,required"`
+}
+type X struct {
+	X string `vflow:"10,required"`
+}
+
+type tAllAttributeTypes struct {
+	S    string  `vflow:"1"`
+	Sp   *string `vflow:"2"`
+	U32  uint32  `vflow:"3"`
+	U32p *uint32 `vflow:"4"`
+	U64  uint64  `vflow:"5"`
+	I32  int32   `vflow:"6"`
+	I64  int64   `vflow:"7"`
+	I64p *int64  `vflow:"8"`
+}
+
+const (
+	recordTypeNestedEmbedded uint32 = iota + 88801
+	recordTypeAllAttributeTypes
+)
+
+func init() {
+	encoding.MustRegisterRecord(recordTypeNestedEmbedded, tNestedEmbedded{})
+	encoding.MustRegisterRecord(recordTypeAllAttributeTypes, tAllAttributeTypes{})
+}
+
+var decoderTests = []struct {
+	CaseName
+	In       map[any]any
+	ErrorMsg string
+	Out      interface{}
+	Golden   bool
+}{
+	{
+		CaseName: Name(""),
+		In: map[any]any{
+			uint32(0): uint32(1), uint32(1): "routerid", uint32(2): "parentid", uint32(3): uint64(100), uint32(4): uint64(1000),
+			uint32(12): "default", uint32(30): "routername",
+		},
+		Out: vanflow.RouterRecord{
+			BaseRecord: vanflow.NewBase("routerid", time.UnixMicro(100), time.UnixMicro(1000)),
+			Parent:     ptrTo("parentid"),
+			Namespace:  ptrTo("default"),
+			Name:       ptrTo("routername"),
+		},
+		Golden: true,
+	}, {
+		CaseName: Name(""),
+		In: map[any]any{
+			uint32(0): uint32(1), uint32(1): "routerid", uint32(2): "parentid", uint32(3): uint64(100), uint32(4): uint64(1000),
+			uint32(12): "default", uint32(30): "routername",
+		},
+		Out: vanflow.RouterRecord{
+			BaseRecord: vanflow.NewBase("routerid", time.UnixMicro(100), time.UnixMicro(1000)),
+			Parent:     ptrTo("parentid"),
+			Namespace:  ptrTo("default"),
+			Name:       ptrTo("routername"),
+		},
+	}, {
+		CaseName: Name(""),
+		ErrorMsg: "decode error: cannot decode nil record attribute set",
+	}, {
+		CaseName: Name(""),
+		In:       func() map[any]any { var nilMap map[any]any; return nilMap }(),
+		ErrorMsg: "decode error: cannot decode nil record attribute set",
+	}, {
+		CaseName: Name(""),
+		In: map[any]any{
+			uint32(0): uint32(404_404), uint32(12): "unknown",
+		},
+		ErrorMsg: "decode error: unknown record type for 404404",
+	}, {
+		CaseName: Name(""),
+		In: map[any]any{
+			uint32(0): uint32(404_404), 12: "unknown",
+		},
+		ErrorMsg: "decode error: unknown record type for 404404",
+	}, {
+		CaseName: Name(""),
+		In: map[any]any{
+			int(0): uint32(404_404), 12: "unknown",
+		},
+		ErrorMsg: "decode error: record type attribute not present",
+	}, {
+		CaseName: Name(""),
+		In: map[any]any{
+			uint32(12): "unknown",
+		},
+		ErrorMsg: "decode error: record type attribute not present",
+	}, {
+		CaseName: Name(""),
+		In: map[any]any{
+			uint32(0): "unknown",
+		},
+		ErrorMsg: `decode error: unexpected type for record type attribute "string"`,
+	}, {
+		CaseName: Name(""),
+		In: map[any]any{
+			uint32(0): uint32(1),
+		},
+		ErrorMsg: `decode error: record attribute set missing required field "ID"`,
+	}, {
+		CaseName: Name(""),
+		In: map[any]any{
+			uint32(0): uint32(1), int64(1): "incorrect key type",
+		},
+		ErrorMsg: `decode error: record attribute set missing required field "ID"`,
+	}, {
+		CaseName: Name(""),
+		In: map[any]any{
+			uint32(0): recordTypeRecordAttributeEncodeDecode, uint32(99): "OKAY", uint32(100): "OKAY",
+		},
+		Out:    tRecordAttributeEncodeDecode{A: ptrTo(MagicBool(true)), B: MagicBool(true)},
+		Golden: true,
+	}, {
+		CaseName: Name(""),
+		In: map[any]any{
+			uint32(0): recordTypeRecordAttributeEncodeDecode,
+		},
+		Out: tRecordAttributeEncodeDecode{},
+	}, {
+		CaseName: Name(""),
+		In: map[any]any{
+			uint32(0): uint32(1), uint32(1): "routerid",
+			uint32(4): uint64(math.MaxInt64) + 1,
+		},
+		ErrorMsg: `decode error: error decoding field "EndTime": time too far in future for internal representation: 0x8000000000000000`,
+	}, {
+		CaseName: Name(""),
+		In: map[any]any{
+			uint32(0): recordTypeNestedEmbedded, uint32(1): uint32(1), uint32(10): "ten", uint32(100): "100",
+		},
+		Out: tNestedEmbedded{Root: 1, A: A{B: B{B: "100"}, X: X{X: "ten"}}},
+	}, {
+		CaseName: Name(""),
+		In: map[any]any{
+			uint32(0): uint32(1), uint32(1): "routerid", uint32(3): uint64(100),
+		},
+		Out: vanflow.RouterRecord{
+			BaseRecord: vanflow.NewBase("routerid", time.UnixMicro(100)),
+		},
+		Golden: true,
+	}, {
+		CaseName: Name("all primitive attributes empty"),
+		In: map[any]any{
+			uint32(0): recordTypeAllAttributeTypes,
+		},
+		Out:    tAllAttributeTypes{},
+		Golden: true,
+	}, {
+		CaseName: Name("all primitive attributes full"),
+		In: map[any]any{
+			uint32(0): recordTypeAllAttributeTypes,
+			uint32(1): "str", uint32(2): "strP",
+			uint32(3): uint32(10), uint32(4): uint32(10),
+			uint32(5): uint64(20),
+			uint32(6): int32(20),
+			uint32(7): int64(30), uint32(8): int64(30),
+		},
+		Out: tAllAttributeTypes{
+			S: "str", Sp: ptrTo("strP"), U32: 10, U32p: ptrTo(uint32(10)), I32: 20, U64: 20, I64: 30, I64p: ptrTo(int64(30)),
+		},
+		Golden: true,
+	},
+}
+
+func TestDecode(t *testing.T) {
+	for _, tc := range decoderTests {
+		t.Run(tc.Name, func(t *testing.T) {
+			out, err := encoding.Decode(tc.In)
+			if !equalError(err, tc.ErrorMsg) {
+				t.Fatalf("%s: unexpected error. wanted: %q but got: %q", tc.Where, tc.ErrorMsg, err)
+			}
+			if tc.Out == nil {
+				return
+			}
+
+			if !cmp.Equal(tc.Out, out) {
+				t.Fatalf("%s: Decode got: %+v want: %+v\n\n%s", tc.Where, out, tc.Out, cmp.Diff(tc.Out, out))
+			}
+
+			if tc.ErrorMsg != "" || !tc.Golden {
+				return
+			}
+
+			remarshaled, err := encoding.Encode(out)
+			if err != nil {
+				t.Fatalf("%s: unexpected encode error: %q", tc.Where, err)
+			}
+			if !cmp.Equal(tc.In, remarshaled) {
+				t.Fatalf("%s: Encode got: %+v want: %+v\n\n%s", tc.Where, remarshaled, tc.In, cmp.Diff(tc.In, remarshaled))
+			}
+
+		})
+	}
+}
+
+func equalError(e error, s string) bool {
+	if e == nil || s == "" {
+		return e == nil && s == ""
+	}
+	return e.Error() == s
+}
+
+type CaseName struct {
+	Name  string
+	Where CasePos
+}
+
+func Name(name string) CaseName {
+	c := CaseName{Name: name}
+	runtime.Callers(2, c.Where[:])
+	return c
+}
+
+type CasePos [1]uintptr
+
+func (p CasePos) String() string {
+	frames := runtime.CallersFrames(p[:])
+	next, _ := frames.Next()
+	return fmt.Sprintf("%s:%d", path.Base(next.File), next.Line)
+}
diff --git a/pkg/vanflow/encoding/encoder_test.go b/pkg/vanflow/encoding/encoder_test.go
new file mode 100644
index 00000000..078e7c68
--- /dev/null
+++ b/pkg/vanflow/encoding/encoder_test.go
@@ -0,0 +1,186 @@
+package encoding_test
+
+import (
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/skupperproject/skupper/pkg/vanflow"
+	"github.com/skupperproject/skupper/pkg/vanflow/encoding"
+	"gotest.tools/assert"
+)
+
+type tRecordAttributeEncodeDecode struct {
+	A *MagicBool `vflow:"99"`
+	B MagicBool  `vflow:"100"`
+}
+
+type MagicBool bool
+
+func (m MagicBool) EncodeRecordAttribute() (interface{}, error) {
+	if m {
+		return "OKAY", nil
+	}
+	return "ERROR", nil
+}
+
+func (m *MagicBool) DecodeRecordAttribute(obj interface{}) error {
+	if obj.(string) == "OKAY" {
+		(*m) = true
+	}
+	return nil
+}
+
+const (
+	recordTypeUnused uint32 = iota + 99900
+	recordTypeRecordAttributeEncodeDecode
+)
+
+func init() {
+	encoding.MustRegisterRecord(recordTypeRecordAttributeEncodeDecode, tRecordAttributeEncodeDecode{})
+}
+
+func TestMustRegister(t *testing.T) {
+	expectPanic := func(t *testing.T, expected string) func() {
+		return func() {
+			if recovered := recover(); recovered != nil {
+				assert.Check(t, strings.HasPrefix(recovered.(string), expected), "got %q but expected %q", recovered, expected)
+				return
+			}
+			t.Fatal("expected MustRegister to panic")
+		}
+	}
+	t.Run("existing codepoint", func(t *testing.T) {
+		defer expectPanic(t, "cannot register record type struct {} using codepoint 99901: already in use")()
+		encoding.MustRegisterRecord(recordTypeRecordAttributeEncodeDecode, struct{}{})
+	})
+	t.Run("repeated vflow attribute tags", func(t *testing.T) {
+		defer expectPanic(t, `struct field B repeats vflow tag "1" also used by A`)()
+		type Repeat struct {
+			A int64 `vflow:"1"`
+			B int64 `vflow:"1"`
+		}
+		encoding.MustRegisterRecord(recordTypeUnused, Repeat{})
+	})
+	t.Run("invalid tag", func(t *testing.T) {
+		defer expectPanic(t, `vflow struct tag parse error for field A:`)()
+		type Repeat struct {
+			A int64 `vflow:"identity"`
+		}
+		encoding.MustRegisterRecord(recordTypeUnused, Repeat{})
+	})
+	t.Run("invalid type", func(t *testing.T) {
+		defer expectPanic(t, `invalid vflow field encoder for "D": unsupported attribute type "testing.B"`)()
+		type Invalid struct {
+			A testing.B
+			B *testing.B
+			c testing.B `vflow:"3"`
+			D testing.B `vflow:"4"`
+		}
+		encoding.MustRegisterRecord(recordTypeUnused, Invalid{c: testing.B{}})
+	})
+	t.Run("register twice", func(t *testing.T) {
+		defer expectPanic(t, `cannot register same type more than once. type encoding_test.tRecordAttributeEncodeDecode already registered with code`)()
+		encoding.MustRegisterRecord(recordTypeUnused, tRecordAttributeEncodeDecode{})
+	})
+}
+
+func TestMarshal(t *testing.T) {
+	timeA := vanflow.Time{Time: time.UnixMicro(100)}
+	timeB := vanflow.Time{Time: time.UnixMicro(333)}
+	testCases := []struct {
+		Name          string
+		Input         interface{}
+		ExpectErr     bool
+		ExpectedAttrs map[interface{}]interface{}
+	}{
+		{
+			Name:      "missing identity error",
+			Input:     vanflow.SiteRecord{},
+			ExpectErr: true,
+		}, {
+			Name:      "nil",
+			ExpectErr: true,
+		}, {
+			Name:  "basic site",
+			Input: vanflow.SiteRecord{BaseRecord: vanflow.BaseRecord{ID: "test"}},
+			ExpectedAttrs: map[any]any{
+				uint32(0): uint32(0),
+				uint32(1): "test",
+			},
+		}, {
+			Name: "full site",
+			Input: vanflow.SiteRecord{
+				Location: ptrTo("loc"),
+				BaseRecord: vanflow.BaseRecord{
+					ID:        "test",
+					StartTime: ptrTo(timeA),
+					EndTime:   ptrTo(timeB),
+				}},
+			ExpectedAttrs: map[any]any{
+				uint32(0): uint32(0),
+				uint32(1): "test",
+				uint32(3): uint64(100),
+				uint32(4): uint64(333),
+				uint32(9): "loc",
+			},
+		}, {
+			Name: "RecordAttributeMarshaler",
+			Input: ptrTo(tRecordAttributeEncodeDecode{
+				A: ptrTo(MagicBool(true)),
+				B: MagicBool(false),
+			}),
+			ExpectedAttrs: map[any]any{
+				uint32(0):   recordTypeRecordAttributeEncodeDecode, // type
+				uint32(99):  "OKAY",
+				uint32(100): "ERROR",
+			},
+		}, {
+			Name:      "unregistered type",
+			Input:     testing.B{},
+			ExpectErr: true,
+		}, {
+			Name:      "time before epoch",
+			Input:     vanflow.SiteRecord{BaseRecord: vanflow.NewBase("test", time.UnixMicro(-1))},
+			ExpectErr: true,
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.Name, func(t *testing.T) {
+			actual, err := encoding.Encode(tc.Input)
+			if tc.ExpectErr {
+				assert.Check(t, err != nil)
+			} else {
+				assert.Check(t, err)
+				assert.DeepEqual(t, tc.ExpectedAttrs, map[any]any(actual))
+			}
+		})
+	}
+}
+
+func BenchmarkEncodeDecode(b *testing.B) {
+	timeA := vanflow.Time{Time: time.UnixMicro(100)}
+	timeB := vanflow.Time{Time: time.UnixMicro(333)}
+	exemplar := vanflow.SiteRecord{
+		Location: ptrTo("loc"),
+		Platform: ptrTo("kubernetes"),
+		Name:     ptrTo("test"),
+		BaseRecord: vanflow.BaseRecord{
+			ID:        "test",
+			StartTime: ptrTo(timeA),
+			EndTime:   ptrTo(timeB),
+		},
+	}
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		attrs, _ := encoding.Encode(exemplar)
+		_, err := encoding.Decode(attrs)
+		if err != nil {
+			b.Fatal(err)
+		}
+	}
+}
+
+func ptrTo[T any](t T) *T {
+	return &t
+}
diff --git a/pkg/vanflow/eventsource/client_test.go b/pkg/vanflow/eventsource/client_test.go
new file mode 100644
index 00000000..b7bb6624
--- /dev/null
+++ b/pkg/vanflow/eventsource/client_test.go
@@ -0,0 +1,280 @@
+package eventsource
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"testing"
+	"time"
+
+	amqp "github.com/Azure/go-amqp"
+	"github.com/skupperproject/skupper/pkg/vanflow"
+	"github.com/skupperproject/skupper/pkg/vanflow/session"
+	"gotest.tools/assert"
+)
+
+func TestClient(t *testing.T) {
+	t.Parallel()
+	tstCtx, tstCancel := context.WithCancel(context.Background())
+	defer tstCancel()
+	factory, rtt := requireContainers(t)
+	ctr, tstCtr := factory.Create(), factory.Create()
+	ctr.Start(tstCtx)
+	tstCtr.Start(tstCtx)
+
+	clientID := uniqueSuffix("test")
+	client := NewClient(ctr, ClientOptions{Source: Info{
+		ID:      clientID,
+		Address: mcsfe(clientID),
+	}})
+	heartbeats := make(chan vanflow.HeartbeatMessage, 8)
+	records := make(chan vanflow.RecordMessage, 8)
+	client.OnHeartbeat(func(m vanflow.HeartbeatMessage) { heartbeats <- m })
+	client.OnRecord(func(m vanflow.RecordMessage) { records <- m })
+
+	sender := tstCtr.NewSender(mcsfe(clientID), session.SenderOptions{})
+	assert.Check(t, client.Listen(tstCtx, FromSourceAddress()))
+
+	heartbeat := vanflow.HeartbeatMessage{
+		Identity: clientID, Version: 1, Now: 22,
+		MessageProps: vanflow.MessageProps{
+			To:      mcsfe(clientID),
+			Subject: "HEARTBEAT",
+		},
+	}
+	initRetryTimer := time.After(250 * rtt)
+	for i := 0; i < 10; i++ {
+		sender.Send(tstCtx, heartbeat.Encode())
+		select {
+		case actual := <-heartbeats:
+			initRetryTimer = nil
+			assert.DeepEqual(t, actual, heartbeat)
+		case <-initRetryTimer:
+			t.Log("retrying heartbeat")
+			initRetryTimer = nil
+		}
+		heartbeat.Now++
+	}
+	record := vanflow.RecordMessage{
+		MessageProps: vanflow.MessageProps{
+			To:      mcsfe(clientID),
+			Subject: "RECORD",
+		},
+	}
+	for i := 0; i < 10; i++ {
+		msg, err := record.Encode()
+		assert.Check(t, err)
+		sender.Send(tstCtx, msg)
+		actual := <-records
+		assert.DeepEqual(t, actual, record)
+		name := fmt.Sprintf("router-%d", i)
+		record.Records = append(record.Records, vanflow.RouterRecord{BaseRecord: vanflow.BaseRecord{ID: name}})
+	}
+
+	closed := make(chan struct{})
+	go func() {
+		defer close(closed)
+		client.Close()
+	}()
+	select {
+	case <-closed: //okay
+	case <-time.After(500 * rtt):
+		t.Error("expected client.Close() to promptly return")
+	}
+
+	msg, err := record.Encode()
+	assert.Check(t, err)
+	sender.Send(tstCtx, msg)
+	select {
+	case <-time.After(100 * time.Millisecond): //okay
+	case <-records:
+		t.Error("expected client to stop handling records after close called")
+	}
+
+}
+
+func TestClientFlush(t *testing.T) {
+	factory, rtt := requireContainers(t)
+	ctr, tstCtr := factory.Create(), factory.Create()
+	ctr.Start(context.Background())
+	tstCtr.Start(context.Background())
+
+	testSuffix := uniqueSuffix("")
+	testCases := []struct {
+		ClientName string
+		When       func(t *testing.T, ctx context.Context, client *Client)
+		Expect     func(t *testing.T, ctx context.Context, flushMsg <-chan *amqp.Message)
+	}{
+		{
+			ClientName: "flush" + testSuffix,
+			When: func(t *testing.T, ctx context.Context, client *Client) {
+				assert.Check(t, client.SendFlush(ctx))
+			},
+			Expect: func(t *testing.T, ctx context.Context, flushMsg <-chan *amqp.Message) {
+				select {
+				case <-time.After(rtt * 5):
+					t.Errorf("expected flush message")
+				case msg := <-flushMsg:
+					assert.Equal(t, *msg.Properties.Subject, "FLUSH")
+				}
+			},
+		}, {
+			ClientName: "noop",
+			When: func(t *testing.T, ctx context.Context, client *Client) {
+			},
+			Expect: func(t *testing.T, ctx context.Context, flushMsg <-chan *amqp.Message) {
+				select {
+				case <-time.After(rtt * 2):
+				case msg := <-flushMsg:
+					t.Errorf("unexpected flush message: %v", msg)
+				}
+			},
+		}, {
+			ClientName: "flush-on-first-message" + testSuffix,
+			When: func(t *testing.T, ctx context.Context, client *Client) {
+				sender := tstCtr.NewSender(mcsfe("flush-on-first-message")+testSuffix, session.SenderOptions{})
+				go sendHeartbeatMessagesTo(t, ctx, sender)
+				assert.Check(t, client.Listen(ctx, FromSourceAddress()))
+				assert.Check(t, FlushOnFirstMessage(ctx, client))
+			},
+			Expect: func(t *testing.T, ctx context.Context, flushMsg <-chan *amqp.Message) {
+				select {
+				case <-time.After(rtt * 5):
+					t.Errorf("expected flush message")
+				case msg := <-flushMsg:
+					assert.Equal(t, *msg.Properties.Subject, "FLUSH")
+				}
+			},
+		}, {
+			ClientName: "flush-on-first-message-timeout" + testSuffix,
+			When: func(t *testing.T, ctx context.Context, client *Client) {
+				flushCtx, cancel := context.WithTimeout(ctx, rtt*10)
+				defer cancel()
+				err := FlushOnFirstMessage(flushCtx, client)
+				assert.Assert(t, err != nil)
+			},
+			Expect: func(t *testing.T, ctx context.Context, flushMsg <-chan *amqp.Message) {
+				select {
+				case <-time.After(rtt * 2):
+				case msg := <-flushMsg:
+					t.Errorf("unexpected flush message: %v", msg)
+				}
+			},
+		},
+	}
+	for _, _tc := range testCases {
+		tc := _tc
+		t.Run(tc.ClientName, func(t *testing.T) {
+			t.Parallel()
+			tstCtx, tstCancel := context.WithCancel(context.Background())
+			defer tstCancel()
+			client := NewClient(ctr, ClientOptions{Source: Info{
+				ID:      tc.ClientName,
+				Address: mcsfe(tc.ClientName),
+				Direct:  sfe(tc.ClientName),
+			}})
+			receiver := tstCtr.NewReceiver(sfe(tc.ClientName), session.ReceiverOptions{})
+
+			flush := make(chan *amqp.Message)
+			go func() {
+				for {
+					msg, err := receiver.Next(tstCtx)
+					if tstCtx.Err() != nil {
+						return
+					}
+					assert.Check(t, err)
+					receiver.Accept(tstCtx, msg)
+					flush <- msg
+				}
+			}()
+			tc.When(t, tstCtx, client)
+			tc.Expect(t, tstCtx, flush)
+		})
+	}
+
+}
+
+func sendHeartbeatMessagesTo(t *testing.T, ctx context.Context, sender session.Sender) {
+	t.Helper()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(time.Millisecond):
+			err := sender.Send(ctx, vanflow.HeartbeatMessage{}.Encode())
+			if ctx.Err() != nil {
+				return
+			}
+			assert.Check(t, err)
+		}
+	}
+}
+
+func requireContainers(t *testing.T) (session.ContainerFactory, time.Duration) {
+	t.Helper()
+	errNotSet := errors.New("SKUPPER_ROUTER_AMQP_ADDRESS environment variable not present")
+	factory, err := func() (session.ContainerFactory, error) {
+		routerAddress := os.Getenv("SKUPPER_ROUTER_AMQP_ADDRESS")
+		if testing.Short() || routerAddress == "" {
+			return nil, errNotSet
+		}
+		setupCtx, cancel := context.WithTimeout(context.Background(), time.Second*3)
+		defer cancel()
+		setupCtr := session.NewContainer(routerAddress, session.ContainerConfig{})
+		setupCtr.Start(setupCtx)
+		pingAddr := uniqueSuffix("ping")
+		ping := setupCtr.NewSender(pingAddr, session.SenderOptions{})
+		pong := setupCtr.NewReceiver(pingAddr, session.ReceiverOptions{})
+
+		sendDone := make(chan error)
+		go func() {
+			defer close(sendDone)
+			sendDone <- ping.Send(setupCtx, amqp.NewMessage([]byte("PING")))
+		}()
+		msg, err := pong.Next(setupCtx)
+		if err != nil {
+			return nil, fmt.Errorf("qdr receive failed: %s: %s", routerAddress, err)
+		}
+		pong.Accept(setupCtx, msg)
+		if err := <-sendDone; err != nil {
+			return nil, fmt.Errorf("qdr send failed: %s: %s", routerAddress, err)
+		}
+		ping.Close(setupCtx)
+		pong.Close(setupCtx)
+
+		factory := session.NewContainerFactory(routerAddress,
+			session.ContainerConfig{
+				ContainerID: uniqueSuffix("eventsourcetest"),
+			})
+		return factory, nil
+	}()
+	if err != nil {
+		if testing.Short() || err == errNotSet {
+			return session.NewMockContainerFactory(), time.Millisecond * 2
+		}
+		t.Fatalf("failed to setup tests: %v", err)
+	}
+
+	return factory, time.Millisecond * 100
+}
+
+func uniqueSuffix(prefix string) string {
+	var salt [8]byte
+	io.ReadFull(rand.Reader, salt[:])
+	out := bytes.NewBuffer([]byte(prefix))
+	out.WriteByte('-')
+	hex.NewEncoder(out).Write(salt[:])
+	return out.String()
+}
+
+func mcsfe(id string) string {
+	return fmt.Sprintf("mc/sfe.%s", id)
+}
+func sfe(id string) string {
+	return fmt.Sprintf("sfe.%s", id)
+}
diff --git a/pkg/vanflow/eventsource/discovery_test.go b/pkg/vanflow/eventsource/discovery_test.go
new file mode 100644
index 00000000..df047322
--- /dev/null
+++ b/pkg/vanflow/eventsource/discovery_test.go
@@ -0,0 +1,252 @@
+package eventsource
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/skupperproject/skupper/pkg/vanflow"
+	"github.com/skupperproject/skupper/pkg/vanflow/session"
+	"gotest.tools/assert"
+	"gotest.tools/poll"
+)
+
+func TestDiscoveryBasic(t *testing.T) {
+	t.Parallel()
+	tstCtx, tstCancel := context.WithCancel(context.Background())
+	defer tstCancel()
+	factory, rtt := requireContainers(t)
+	ctr, tstCtr := factory.Create(), factory.Create()
+	ctr.Start(tstCtx)
+	tstCtr.Start(tstCtx)
+
+	beaconAddress := mcsfe(uniqueSuffix("all"))
+	discovery := NewDiscovery(ctr, DiscoveryOptions{BeaconAddress: beaconAddress})
+
+	discoveredOut := make(chan Info, 8)
+	forgottenOut := make(chan Info, 8)
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		discovery.Run(tstCtx, DiscoveryHandlers{
+			Discovered: func(info Info) {
+				discoveredOut <- info
+			},
+			Forgotten: func(info Info) {
+				forgottenOut <- info
+			},
+		})
+	}()
+
+	tstSender := tstCtr.NewSender(beaconAddress, session.SenderOptions{})
+
+	testSuffix := uniqueSuffix("")
+	sourceAID, sourceBID := "a"+testSuffix, "b"+testSuffix
+	beaconA := fixtureBeaconFor(sourceAID, "ROUTER")
+	beaconB := fixtureBeaconFor(sourceBID, "CONTROLLER")
+
+	tstSender.Send(tstCtx, beaconA.Encode())
+	tstSender.Send(tstCtx, beaconB.Encode())
+
+	// when listeners are not yet present the router can drop beacon messages
+	// retry sending initial beacons once
+	retryOnceAfter := time.Now().Add(500 * rtt)
+	// wait for discovery.List to return two sources
+	poll.WaitOn(t,
+		func(t poll.LogT) poll.Result {
+			actual, desired := len(discovery.List()), 2
+			if actual == desired {
+				return poll.Success()
+			}
+			if time.Now().After(retryOnceAfter) {
+				tstSender.Send(tstCtx, beaconA.Encode())
+				tstSender.Send(tstCtx, beaconB.Encode())
+				retryOnceAfter = retryOnceAfter.Add(time.Hour)
+			}
+			return poll.Continue("number of event sources is %d, not %d", actual, desired)
+		}, poll.WithTimeout(3000*rtt),
+	)
+
+	// expect two events
+	eventA := <-discoveredOut
+	eventB := <-discoveredOut
+	if eventA.ID == sourceBID {
+		eventA, eventB = eventB, eventA
+	}
+
+	sourceA, ok := discovery.Get(sourceAID)
+	assert.Check(t, ok)
+	sourceB, ok := discovery.Get(sourceBID)
+	assert.Check(t, ok)
+
+	assert.DeepEqual(t, sourceA, eventA, cmpopts.IgnoreFields(Info{}, "LastSeen"))
+	assert.DeepEqual(t, sourceB, eventB, cmpopts.IgnoreFields(Info{}, "LastSeen"))
+
+	tstSender.Send(tstCtx, beaconA.Encode())
+	tstSender.Send(tstCtx, beaconA.Encode())
+	tstSender.Send(tstCtx, beaconA.Encode())
+	tstSender.Send(tstCtx, beaconA.Encode())
+	tstSender.Send(tstCtx, beaconB.Encode())
+	tstSender.Send(tstCtx, beaconA.Encode())
+
+	// wait for LastSeen to update
+	poll.WaitOn(t,
+		func(t poll.LogT) poll.Result {
+			presentA, ok := discovery.Get(sourceAID)
+			if !ok {
+				return poll.Error(fmt.Errorf("error getting source 'a'"))
+			}
+			presentB, ok := discovery.Get(sourceBID)
+			if !ok {
+				return poll.Error(fmt.Errorf("error getting source 'b'"))
+			}
+			prevA, currentA := eventA.LastSeen, presentA.LastSeen
+			prevB, currentB := eventB.LastSeen, presentB.LastSeen
+			if currentA.After(prevA) && currentB.After(prevB) {
+				return poll.Success()
+			}
+			return poll.Continue("waiting for lastseen to advance")
+		}, poll.WithTimeout(300*rtt),
+	)
+	assert.Check(t, len(discoveredOut) == 0, "expected no new discovery events after subsequent beacons")
+	assert.Check(t, len(forgottenOut) == 0, "expected no new forgotten events after subsequent beacons")
+
+	assert.Check(t, !discovery.Forget("c"), "expected to ignore call to Forget for unknown id")
+	assert.Check(t, len(forgottenOut) == 0, "expected no new events after invalid call to Forget")
+
+	assert.Check(t, discovery.Forget(sourceAID), "expected ok to forget event source 'a'")
+	// wait for discovery.List to return only one source
+	poll.WaitOn(t,
+		func(t poll.LogT) poll.Result {
+			actual, desired := len(discovery.List()), 1
+			if actual == desired {
+				return poll.Success()
+			}
+			return poll.Continue("number of event sources is %d, not %d", actual, desired)
+		}, poll.WithTimeout(1000*rtt),
+	)
+
+	// expect one event
+	eventDelete := <-forgottenOut
+	assert.Check(t, eventDelete.ID == sourceAID)
+	assert.Equal(t, len(discovery.List()), 1)
+	_, ok = discovery.Get(sourceAID)
+	assert.Check(t, !ok, "expected Get on forgotten ID to return not ok")
+
+	tstCancel()
+	select {
+	case <-time.After(rtt * 500):
+		t.Error("expected discovery.Run to finish after cancelling context")
+	case <-done: // okay
+	}
+}
+
+func TestDiscoveryWatch(t *testing.T) {
+	t.Parallel()
+	tstCtx, tstCancel := context.WithCancel(context.Background())
+	defer tstCancel()
+	factory, rtt := requireContainers(t)
+	ctr, tstCtr := factory.Create(), factory.Create()
+	ctr.Start(tstCtx)
+	tstCtr.Start(tstCtx)
+
+	beaconAddress := mcsfe(uniqueSuffix("all"))
+	discovery := NewDiscovery(ctr, DiscoveryOptions{BeaconAddress: beaconAddress})
+
+	discoveredOut := make(chan Info, 8)
+	forgottenOut := make(chan Info, 8)
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		discovery.Run(tstCtx, DiscoveryHandlers{
+			Discovered: func(info Info) {
+				discoveredOut <- info
+			},
+			Forgotten: func(info Info) {
+				forgottenOut <- info
+			},
+		})
+	}()
+
+	sourceAID := uniqueSuffix("a")
+
+	beaconSender := tstCtr.NewSender(beaconAddress, session.SenderOptions{})
+	heartbeatSender := tstCtr.NewSender(mcsfe(sourceAID), session.SenderOptions{})
+	// continually send heartbeats for source a
+	go func() {
+		heartbeat := vanflow.HeartbeatMessage{
+			Version:      1,
+			Now:          1000,
+			Identity:     sourceAID,
+			MessageProps: vanflow.MessageProps{To: mcsfe(sourceAID)},
+		}
+		for {
+			time.Sleep(rtt)
+			heartbeatSender.Send(tstCtx, heartbeat.Encode())
+			heartbeat.Now++
+		}
+	}()
+
+	// send a beacon for router a and await the discovery event
+	beaconA := fixtureBeaconFor(sourceAID, "ROUTER")
+
+	assert.Check(t, beaconSender.Send(tstCtx, beaconA.Encode()))
+	var event Info
+	select {
+	case out := <-discoveredOut:
+		event = out
+	case <-time.After(10 * rtt):
+		t.Log("retrying beacon") //
+		assert.Check(t, beaconSender.Send(tstCtx, beaconA.Encode()))
+		event = <-discoveredOut
+	}
+
+	client := NewClient(ctr, ClientOptions{Source: event})
+	// start a new watched client and begin listening
+	err := discovery.NewWatchClient(tstCtx, WatchConfig{
+		Client:                  client,
+		ID:                      event.ID,
+		Timeout:                 rtt * 10,
+		GracePeriod:             time.Second,
+		DiscoveryUpdateInterval: rtt * 2,
+	})
+	assert.Check(t, err)
+
+	listenCtx, listenCancel := context.WithCancel(tstCtx)
+	client.Listen(listenCtx, FromSourceAddress())
+
+	poll.WaitOn(t,
+		func(t poll.LogT) poll.Result {
+			present, ok := discovery.Get(sourceAID)
+			if !ok {
+				return poll.Error(fmt.Errorf("event source 'a' forgotten"))
+			}
+			prev, current := event.LastSeen, present.LastSeen
+			if current.After(prev) {
+				return poll.Success()
+			}
+			return poll.Continue("waiting for lastseen to advance")
+		}, poll.WithDelay(rtt), poll.WithTimeout(300*rtt),
+	)
+
+	listenCancel()
+
+	select {
+	case event := <-forgottenOut:
+		assert.Check(t, event.ID == sourceAID)
+	case <-time.After(80 * rtt):
+		t.Error("expected source to be forgotten after starting watch client with no activity")
+	}
+}
+
+func fixtureBeaconFor(id string, source string) vanflow.BeaconMessage {
+	return vanflow.BeaconMessage{
+		Version:    1,
+		SourceType: source,
+		Address:    fmt.Sprintf("mc/sfe.%s", id),
+		Direct:     fmt.Sprintf("sfe.%s", id),
+		Identity:   id,
+	}
+}
diff --git a/pkg/vanflow/eventsource/manager_test.go b/pkg/vanflow/eventsource/manager_test.go
new file mode 100644
index 00000000..e4ff5279
--- /dev/null
+++ b/pkg/vanflow/eventsource/manager_test.go
@@ -0,0 +1,99 @@
+package eventsource
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/skupperproject/skupper/pkg/vanflow"
+	"github.com/skupperproject/skupper/pkg/vanflow/store"
+	"gotest.tools/poll"
+)
+
+func TestManagerClient(t *testing.T) {
+	t.Parallel()
+	tstCtx, tstCancel := context.WithCancel(context.Background())
+	defer tstCancel()
+
+	sourceID := uniqueSuffix("test-event-source-manager")
+	factory, rtt := requireContainers(t)
+	ctr := factory.Create()
+	ctr.Start(tstCtx)
+	ctrClient := factory.Create()
+	ctrClient.Start(tstCtx)
+
+	discovery := NewDiscovery(ctrClient, DiscoveryOptions{})
+	go discovery.Run(tstCtx, DiscoveryHandlers{})
+
+	sourceRef := store.SourceRef{ID: sourceID}
+	source := Info{ID: sourceID, Type: "test-event-source", Address: mcsfe(sourceID), Direct: sfe(sourceID)}
+	logStor := store.NewSyncMapStore(store.SyncMapStoreConfig{})
+	listenerStor := store.NewSyncMapStore(store.SyncMapStoreConfig{})
+	for i := 0; i < 64; i++ {
+		logStor.Add(vanflow.LogRecord{BaseRecord: vanflow.NewBase(fmt.Sprintf("log-%d", i))}, sourceRef)
+		listenerStor.Add(vanflow.ListenerRecord{BaseRecord: vanflow.NewBase(fmt.Sprintf("listener-%d", i))}, sourceRef)
+	}
+	manager := NewManager(ctr, ManagerConfig{
+		Source:            source,
+		Stores:            []store.Interface{logStor, listenerStor},
+		HeartbeatInterval: rtt * 10,
+		BeaconInterval:    rtt * 50,
+	})
+	go manager.Run(tstCtx)
+
+	clientStor := store.NewSyncMapStore(store.SyncMapStoreConfig{})
+	logsOnly := RecordStoreRouter{
+		Source: sourceRef,
+		Stores: RecordStoreMap{vanflow.LogRecord{}.GetTypeMeta().String(): clientStor},
+	}
+
+	client := NewClient(ctrClient, ClientOptions{Source: source})
+	client.OnRecord(logsOnly.Route)
+	client.Listen(tstCtx, FromSourceAddress())
+	defer client.Close()
+	t.Run("wait to flush", func(t *testing.T) {
+		flushCtx, cancel := context.WithTimeout(tstCtx, rtt*100)
+		defer cancel()
+		if err := FlushOnFirstMessage(flushCtx, client); err != nil {
+			t.Errorf("expected manager to be discovered by the client: %s", err)
+		}
+	})
+
+	t.Run("will syncronize records", func(t *testing.T) {
+		poll.WaitOn(t, func(t poll.LogT) poll.Result {
+			actual := clientStor.List()
+			expected := logStor.List()
+			if !cmp.Equal(actual, expected, ignoreLastUpdateAndOrder...) {
+				return poll.Continue("waiting for all records to be synced: %s", cmp.Diff(actual, expected, ignoreLastUpdateAndOrder...))
+			}
+			return poll.Success()
+		}, poll.WithDelay(rtt), poll.WithTimeout(300*rtt))
+	})
+
+	t.Run("can publish updates", func(t *testing.T) {
+		prevEntry, _ := logStor.Get("log-1")
+		updated := prevEntry.Record.(vanflow.LogRecord)
+		txt := "updated txt"
+		updated.LogText = &txt
+		manager.PublishUpdate(RecordUpdate{Prev: prevEntry.Record, Curr: updated})
+		poll.WaitOn(t, func(t poll.LogT) poll.Result {
+			actual, _ := clientStor.Get("log-1")
+			expected := updated
+			if !cmp.Equal(actual.Record, expected) {
+				return poll.Continue("waiting for record '1' to be updated: %s", cmp.Diff(actual.Record, expected))
+			}
+			return poll.Success()
+		}, poll.WithDelay(rtt), poll.WithTimeout(100*rtt))
+	})
+
+}
+
+var ignoreLastUpdateAndOrder = []cmp.Option{
+	cmpopts.IgnoreFields(store.Metadata{}, "LastUpdate"),
+	cmpopts.SortSlices(func(a, b store.Entry) bool {
+		return strings.Compare(a.Record.Identity(), b.Record.Identity()) < 0
+	}),
+}
diff --git a/pkg/vanflow/messages_test.go b/pkg/vanflow/messages_test.go
new file mode 100644
index 00000000..dc4f7b6f
--- /dev/null
+++ b/pkg/vanflow/messages_test.go
@@ -0,0 +1,188 @@
+package vanflow
+
+import (
+	"testing"
+	"time"
+
+	amqp "github.com/Azure/go-amqp"
+	"gotest.tools/assert"
+)
+
+func TestDecode(t *testing.T) {
+	testCases := []struct {
+		Name  string
+		In    *amqp.Message
+		Error string
+		Out   any
+	}{
+		{
+			Name: "beacon",
+			In:   BeaconMessage{}.Encode(),
+			Out: BeaconMessage{MessageProps: MessageProps{
+				Subject: "BEACON",
+				To:      "mc/sfe.all",
+			}},
+		}, {
+			Name: "heartbeat",
+			In:   HeartbeatMessage{Identity: "a"}.Encode(),
+			Out: HeartbeatMessage{Identity: "a", MessageProps: MessageProps{
+				Subject: "HEARTBEAT",
+				To:      "mc/sfe.a",
+			}},
+		}, {
+			Name: "flush",
+			In:   FlushMessage{}.Encode(),
+			Out:  FlushMessage{MessageProps: MessageProps{Subject: "FLUSH"}},
+		}, {
+			Name:  "nil",
+			In:    nil,
+			Error: "cannot decode message",
+		}, {
+			Name:  "nil props",
+			In:    new(amqp.Message),
+			Error: "cannot decode message",
+		}, {
+			Name:  "no subject",
+			In:    &amqp.Message{Properties: &amqp.MessageProperties{}},
+			Error: "cannot decode message",
+		}, {
+			Name:  "unknown subject",
+			In:    &amqp.Message{Properties: &amqp.MessageProperties{Subject: ptrTo("A")}},
+			Error: "cannot decode message",
+		}, {
+			Name: "emtpy record",
+			In:   func() *amqp.Message { m, _ := RecordMessage{}.Encode(); return m }(),
+			Out:  RecordMessage{MessageProps: MessageProps{Subject: "RECORD"}},
+		}, {
+			Name: "record",
+			In: func() *amqp.Message {
+				r := RecordMessage{
+					Records: []Record{
+						SiteRecord{
+							BaseRecord: NewBase("1", time.UnixMicro(222)),
+						},
+						RouterRecord{BaseRecord: BaseRecord{ID: "1"}},
+						LinkRecord{
+							BaseRecord: BaseRecord{ID: "1"},
+							LinkCost:   ptrTo[uint64](1),
+						},
+					},
+				}
+				m, err := r.Encode()
+				if err != nil {
+					t.Fatal(err)
+				}
+				return m
+			}(),
+			Out: RecordMessage{
+				MessageProps: MessageProps{Subject: "RECORD"},
+				Records: []Record{
+					SiteRecord{
+						BaseRecord: NewBase("1", time.UnixMicro(222)),
+					},
+					RouterRecord{BaseRecord: BaseRecord{ID: "1"}},
+					LinkRecord{
+						BaseRecord: BaseRecord{ID: "1"},
+						LinkCost:   ptrTo[uint64](1),
+					},
+				},
+			},
+		}, {
+			Name: "bad record val",
+			In: func() *amqp.Message {
+				m, _ := RecordMessage{}.Encode()
+				m.Value = []byte("testing")
+				return m
+			}(),
+			Error: "unexpected type for message Value",
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.Name, func(t *testing.T) {
+			actual, err := Decode(tc.In)
+			if tc.Error != "" {
+				assert.ErrorContains(t, err, tc.Error)
+				return
+			}
+			assert.DeepEqual(t, actual, tc.Out)
+		})
+	}
+}
+func TestBeaconMessage(t *testing.T) {
+	original := BeaconMessage{
+		Version:    99,
+		SourceType: "source",
+		Address:    "address",
+		Direct:     "direct",
+		Identity:   "id",
+	}
+	msg := original.Encode()
+	assert.Equal(t, *msg.Properties.Subject, "BEACON")
+	assert.Equal(t, *msg.Properties.To, "mc/sfe.all")
+	dup := DecodeBeacon(msg)
+	dup.MessageProps = MessageProps{}
+	assert.DeepEqual(t, original, dup)
+}
+
+func TestHeartbeatMessage(t *testing.T) {
+	original := HeartbeatMessage{
+		MessageProps: MessageProps{
+			To:      "ignored",
+			Subject: "ignored",
+			ReplyTo: "ignored",
+		},
+		Identity: "myid",
+		Version:  1,
+		Now:      9999,
+	}
+	msgPropsExpected := MessageProps{
+		To:      "mc/sfe.myid",
+		Subject: "HEARTBEAT",
+		ReplyTo: "",
+	}
+	msg := original.Encode()
+	dup := DecodeHeartbeat(msg)
+	original.MessageProps = msgPropsExpected
+	assert.DeepEqual(t, original, dup)
+}
+
+func TestFlushMessage(t *testing.T) {
+	original := FlushMessage{
+		MessageProps{
+			To:      "flushaddr",
+			Subject: "ignored",
+			ReplyTo: "flushreply",
+		},
+	}
+	msgPropsExpected := MessageProps{
+		To:      "flushaddr",
+		Subject: "FLUSH",
+		ReplyTo: "flushreply",
+	}
+	msg := original.Encode()
+	dupe := DecodeFlush(msg)
+	original.MessageProps = msgPropsExpected
+	assert.DeepEqual(t, original, dupe)
+}
+
+func TestRecordMessage(t *testing.T) {
+	original := RecordMessage{
+		MessageProps: MessageProps{
+			To:      "recordaddr",
+			Subject: "ignored",
+			ReplyTo: "ignored",
+		},
+	}
+	msgPropsExpected := MessageProps{
+		Subject: "RECORD",
+		To:      "recordaddr",
+	}
+	msg, err := original.Encode()
+	assert.Check(t, err)
+	dupe, err := DecodeRecord(msg)
+	assert.Check(t, err)
+	original.MessageProps = msgPropsExpected
+	assert.DeepEqual(t, original, dupe)
+}
+
+func ptrTo[T any](obj T) *T { return &obj }
diff --git a/pkg/vanflow/session/container_test.go b/pkg/vanflow/session/container_test.go
new file mode 100644
index 00000000..c96449c4
--- /dev/null
+++ b/pkg/vanflow/session/container_test.go
@@ -0,0 +1,112 @@
+package session
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/Azure/go-amqp"
+	"gotest.tools/assert"
+)
+
+func TestContainerPing(t *testing.T) {
+	testCases := []struct {
+		Name    string
+		Factory func(t *testing.T) ContainerFactory
+	}{
+		{
+			Name: "mock",
+			Factory: func(t *testing.T) ContainerFactory {
+				return NewMockContainerFactory()
+			},
+		}, {
+			Name: "qdr",
+			Factory: func(t *testing.T) ContainerFactory {
+				return containersFromEnv(t)
+			},
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.Name, func(t *testing.T) {
+			factory := tc.Factory(t)
+			cs := factory.Create()
+			cr := factory.Create()
+
+			channel := "ex/" + randomID()
+			ctx := context.Background()
+			cs.Start(ctx)
+			cr.Start(ctx)
+			s := cs.NewSender(channel, SenderOptions{})
+			r := cr.NewReceiver(channel, ReceiverOptions{})
+
+			done := make(chan struct{})
+			go func() {
+				defer close(done)
+				for i := 0; i < 64; i++ {
+					func() {
+						sctx, cancel := context.WithTimeout(ctx, time.Second)
+						defer cancel()
+						err := s.Send(sctx, amqp.NewMessage(
+							[]byte(fmt.Sprintf("ping-%d", i))),
+						)
+						if err != nil {
+							t.Error(err)
+						}
+					}()
+				}
+			}()
+
+			for i := 0; i < 64; i++ {
+				msg, err := r.Next(ctx)
+				if err != nil {
+					t.Error(err)
+				}
+				if err := r.Accept(ctx, msg); err != nil {
+					t.Error(err)
+				}
+			}
+			select {
+			case <-time.After(time.Second * 2):
+				t.Fatal("send timed out")
+			case <-done: // pass
+			}
+
+			assert.Check(t, s.Close(ctx))
+			err := s.Send(ctx, amqp.NewMessage([]byte("closed")))
+			assert.ErrorContains(t, err, "closed")
+		})
+	}
+
+}
+
+func containersFromEnv(t *testing.T) ContainerFactory {
+	t.Helper()
+	if testing.Short() {
+		t.Skip("skipping test that requires external router")
+	}
+	var factory ContainerFactory
+	err := func() error {
+		qdr := os.Getenv("SKUPPER_ROUTER_AMQP_ADDRESS")
+		if qdr == "" {
+			return fmt.Errorf("SKUPPER_ROUTER_AMQP_ADDRESS environment variable not present")
+		}
+		setupCtx, cancel := context.WithTimeout(context.Background(), time.Second*3)
+		defer cancel()
+		conn, err := amqp.Dial(setupCtx, qdr, nil)
+		if err != nil {
+			return fmt.Errorf("could not establish connection to router: %v", err)
+		}
+		conn.Close()
+		factory = NewContainerFactory(qdr, ContainerConfig{
+			ContainerID: "tc/" + randomID(),
+		})
+		return nil
+	}()
+	if err != nil {
+		t.Skipf("skipping test that requires external router: %s", err)
+	}
+
+	return factory
+}
diff --git a/pkg/vanflow/store/syncmap_test.go b/pkg/vanflow/store/syncmap_test.go
new file mode 100644
index 00000000..5cb4ba0b
--- /dev/null
+++ b/pkg/vanflow/store/syncmap_test.go
@@ -0,0 +1,240 @@
+package store
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/skupperproject/skupper/pkg/vanflow"
+)
+
+func TestSyncMapStoreDelete(t *testing.T) {
+	var deleteEvents []Entry
+	stor := NewSyncMapStore(SyncMapStoreConfig{Handlers: EventHandlerFuncs{
+		OnAdd:    func(Entry) { t.Errorf("unexpected call to OnAdd") },
+		OnChange: func(_, _ Entry) { t.Errorf("unexpected call to OnChange") },
+		OnDelete: func(e Entry) { deleteEvents = append(deleteEvents, e) },
+	}})
+
+	source := SourceRef{ID: "test", Version: "0"}
+	r0 := vanflow.LogRecord{BaseRecord: vanflow.NewBase("0", time.Now()), LogText: ptrTo("txt")}
+	initialState := Entry{
+		Record: r0,
+		Metadata: Metadata{
+			Source:     source,
+			LastUpdate: time.Now().Add(-1 * time.Minute),
+		},
+	}
+	stor.(*syncMapStore).Replace([]Entry{initialState})
+
+	deleted, ok := stor.Delete("0")
+	if !ok {
+		t.Error("expected to delete record")
+	}
+	if !cmp.Equal(deleted, initialState) {
+		t.Errorf("expected delete to return previous value: %s", cmp.Diff(deleted, initialState))
+	}
+}
+
+func TestSyncMapStoreAdd(t *testing.T) {
+	var addEvents []Entry
+	stor := NewSyncMapStore(SyncMapStoreConfig{Handlers: EventHandlerFuncs{
+		OnAdd:    func(e Entry) { addEvents = append(addEvents, e) },
+		OnChange: func(_, _ Entry) { t.Errorf("unexpected call to OnChange") },
+		OnDelete: func(Entry) { t.Errorf("unexpected call to OnDelete") },
+	}})
+
+	recordsToAdd := []vanflow.Record{
+		vanflow.LogRecord{BaseRecord: vanflow.NewBase("1")},
+		vanflow.LogRecord{BaseRecord: vanflow.NewBase("2")},
+		vanflow.LogRecord{BaseRecord: vanflow.NewBase("3")},
+	}
+	source := SourceRef{ID: "test", Version: "0"}
+	for _, record := range recordsToAdd {
+		if ok := stor.Add(record, source); !ok {
+			t.Errorf("unexpected failure to add record %v", record)
+		}
+	}
+	for _, record := range recordsToAdd {
+		if ok := stor.Add(record, source); ok {
+			t.Errorf("add record returned ok, but should have failed to re-add record %v", record)
+		}
+	}
+
+	actual := stor.List()
+	expected := make([]Entry, len(recordsToAdd))
+	for i := range recordsToAdd {
+		expected[i] = Entry{Record: recordsToAdd[i], Metadata: Metadata{Source: source}}
+	}
+
+	if !cmp.Equal(actual, expected, ignoreLastUpdateAndOrder...) {
+		t.Errorf("store contents do not match expected: %s", cmp.Diff(actual, expected, ignoreLastUpdateAndOrder...))
+	}
+	if !cmp.Equal(addEvents, expected, ignoreLastUpdateAndOrder...) {
+		t.Errorf("entries from OnAdd handler do not match expected: %s", cmp.Diff(addEvents, expected, ignoreLastUpdateAndOrder...))
+	}
+
+}
+
+func TestSyncMapStoreUpdate(t *testing.T) {
+	source := SourceRef{ID: "test", Version: "0"}
+	var updateEvents []Entry
+	stor := NewSyncMapStore(SyncMapStoreConfig{Handlers: EventHandlerFuncs{
+		OnAdd:    func(Entry) { t.Errorf("unexpected call to OnAdd") },
+		OnChange: func(p, n Entry) { updateEvents = append(updateEvents, p, n) },
+		OnDelete: func(Entry) { t.Errorf("unexpected call to OnDelete") },
+	}})
+
+	r0 := vanflow.LogRecord{BaseRecord: vanflow.NewBase("0")}
+	if ok := stor.Update(r0); ok {
+		t.Errorf("update record returned ok but should have failed to update new record")
+	}
+
+	initialState := Entry{
+		Record: r0,
+		Metadata: Metadata{
+			Source:     source,
+			LastUpdate: time.Now().Add(-1 * time.Minute),
+		},
+	}
+	stor.(*syncMapStore).Replace([]Entry{initialState})
+
+	r0.LogText = ptrTo("test")
+	if ok := stor.Update(r0); !ok {
+		t.Errorf("update record should have returned ok")
+	}
+
+	actual, ok := stor.Get("0")
+	if !ok {
+		t.Fatalf("record not in store")
+	}
+	if !actual.LastUpdate.After(initialState.LastUpdate) {
+		t.Errorf("expected entry.LastUpdate to be updated: delta was %q", actual.LastUpdate.Sub(initialState.LastUpdate))
+	}
+	if !cmp.Equal(r0, actual.Record) {
+		t.Errorf("store contents do not match expected: %s", cmp.Diff(actual.Record, r0))
+	}
+
+	expectedChanges := []Entry{
+		initialState, // prev
+		{Record: r0, Metadata: Metadata{Source: source}}, // next
+	}
+	if !cmp.Equal(updateEvents, expectedChanges, ignoreLastUpdateAndOrder...) {
+		t.Errorf("entries from OnChange handler do not match expected: %s", cmp.Diff(updateEvents, expectedChanges, ignoreLastUpdateAndOrder...))
+	}
+}
+
+func TestSyncMapStorePatch(t *testing.T) {
+	source := SourceRef{ID: "test", Version: "0"}
+	var addEvents []Entry
+	var updateEvents []Entry
+	stor := NewSyncMapStore(SyncMapStoreConfig{Handlers: EventHandlerFuncs{
+		OnAdd:    func(e Entry) { addEvents = append(addEvents, e) },
+		OnChange: func(p, n Entry) { updateEvents = append(updateEvents, p, n) },
+		OnDelete: func(Entry) { t.Errorf("unexpected call to OnDelete") },
+	}})
+
+	startTime := time.Now().Truncate(time.Millisecond)
+	r0 := vanflow.RouterRecord{BaseRecord: vanflow.NewBase("0", startTime), Parent: ptrTo("site")}
+	r0Delta := vanflow.RouterRecord{BaseRecord: vanflow.NewBase("0"), Namespace: ptrTo("ns")}
+	r0Expected := vanflow.RouterRecord{BaseRecord: vanflow.NewBase("0", startTime), Parent: ptrTo("site"), Namespace: ptrTo("ns")}
+
+	initialState := Entry{
+		Record: r0,
+		Metadata: Metadata{
+			Source:     source,
+			LastUpdate: time.Now().Add(-1 * time.Minute),
+		},
+	}
+	stor.(*syncMapStore).Replace([]Entry{initialState})
+
+	stor.Patch(r0, source)      // no change
+	stor.Patch(r0Delta, source) // update r0
+	r1 := vanflow.LogRecord{BaseRecord: vanflow.NewBase("1"), LogText: ptrTo("testI")}
+	stor.Patch(r1, source) // add r1
+	stor.Patch(r1, source) // no change
+
+	actual := stor.List()
+	expected := []Entry{
+		{Record: r0Expected, Metadata: Metadata{Source: source}},
+		{Record: r1, Metadata: Metadata{Source: source}},
+	}
+	if !cmp.Equal(actual, expected, ignoreLastUpdateAndOrder...) {
+		t.Errorf("store contents do not match expected: %s", cmp.Diff(actual, expected, ignoreLastUpdateAndOrder...))
+	}
+
+	expectedAdds := []Entry{
+		{Record: r1, Metadata: Metadata{Source: source}},
+	}
+	if !cmp.Equal(addEvents, expectedAdds, ignoreLastUpdateAndOrder...) {
+		t.Errorf("entries from OnAdd handler do not match expected: %s", cmp.Diff(addEvents, expectedAdds, ignoreLastUpdateAndOrder...))
+	}
+	expectedChanges := []Entry{
+		{Record: r0, Metadata: Metadata{Source: source}},         // prev
+		{Record: r0Expected, Metadata: Metadata{Source: source}}, // next
+	}
+	if !cmp.Equal(updateEvents, expectedChanges, ignoreLastUpdateAndOrder...) {
+		t.Errorf("entries from OnChange handler do not match expected: %s", cmp.Diff(updateEvents, expectedChanges, ignoreLastUpdateAndOrder...))
+	}
+}
+func TestSyncMapStoreIndex(t *testing.T) {
+	stor := NewSyncMapStore(SyncMapStoreConfig{})
+
+	var initialState []Entry
+	for i := 0; i < 8; i++ {
+		source := SourceRef{ID: fmt.Sprint(i)}
+		for c := 0; c < 128; c++ {
+			initialState = append(initialState, Entry{
+				Metadata: Metadata{Source: source},
+				Record:   vanflow.LogRecord{BaseRecord: vanflow.NewBase(fmt.Sprintf("%d-%d", i, c))},
+			})
+		}
+	}
+
+	stor.(*syncMapStore).Replace(initialState)
+
+	items := stor.Index(SourceIndex, Entry{Metadata: Metadata{Source: SourceRef{ID: "dne"}}})
+	if expected, actual := 0, len(items); expected != actual {
+		t.Errorf("expected %d entries for source 'dne' but got %d", expected, actual)
+	}
+
+	items = stor.Index(TypeIndex, Entry{Record: vanflow.LogRecord{}})
+	if expected, actual := initialState, items; !cmp.Equal(expected, actual, ignoreLastUpdateAndOrder...) {
+		t.Errorf("expected type index to return all log records: %s", cmp.Diff(expected, actual, ignoreLastUpdateAndOrder...))
+	}
+
+	items = stor.Index("IndexDoesNotExist", Entry{Metadata: Metadata{Source: SourceRef{ID: "1"}}})
+	if expected, actual := 0, len(items); expected != actual {
+		t.Errorf("expected %d entries for source '1' but got %d", expected, actual)
+	}
+
+	items = stor.Index(SourceIndex, Entry{Metadata: Metadata{Source: SourceRef{ID: "7"}}})
+	if expected, actual := 128, len(items); expected != actual {
+		t.Errorf("expected %d entries for source '7' but got %d", expected, actual)
+	}
+
+	_, ok := stor.Delete(items[0].Record.Identity())
+	if !ok {
+		t.Fatalf("expected delete to succeed")
+	}
+
+	items = stor.Index(SourceIndex, Entry{Metadata: Metadata{Source: SourceRef{ID: "7"}}})
+	if expected, actual := 127, len(items); expected != actual {
+		t.Errorf("expected %d entries for source '7' after deleting one but got %d", expected, actual)
+	}
+
+}
+
+func ptrTo[T any](e T) *T {
+	return &e
+}
+
+var ignoreLastUpdateAndOrder = []cmp.Option{
+	cmpopts.IgnoreFields(Metadata{}, "LastUpdate"),
+	cmpopts.SortSlices(func(a, b Entry) bool {
+		return strings.Compare(a.Record.Identity(), b.Record.Identity()) < 0
+	}),
+}