improvements to the dynamic scoping system

caffix · caffix · commit 3b2e8036e022 · 2026-01-22T20:46:16.000-05:00
diff --git a/engine/dispatcher/dispatcher.go b/engine/dispatcher/dispatcher.go
@@ -103,7 +103,7 @@ func (d *dynamicDispatcher) completedCallback(ede *et.EventDataElement) {
 
 func (d *dynamicDispatcher) DispatchEvent(e *et.Event) error {
 	if e == nil || e.Entity == nil || e.Session == nil {
-		return nil
+		return errors.New("the event cannot be nil and must include the entity and session")
 	}
 
 	// do not schedule the same asset more than once
@@ -129,6 +129,33 @@ func (d *dynamicDispatcher) DispatchEvent(e *et.Event) error {
 	return pool.Dispatch(e)
 }
 
+func (d *dynamicDispatcher) ResubmitEvent(e *et.Event) error {
+	if e == nil || e.Entity == nil || e.Session == nil {
+		return errors.New("the event cannot be nil and must include the entity and session")
+	}
+
+	if !e.Session.Backlog().Has(e.Entity) {
+		return errors.New("the event must already be in the backlog")
+	}
+
+	if err := d.meta.InsertEntry(e.Session.ID().String(), e.Entity.ID, e.Meta); err != nil {
+		return err
+	}
+
+	err := e.Session.Backlog().Enqueue(e.Entity)
+	if err != nil {
+		return err
+	}
+
+	atype := e.Entity.Asset.AssetType()
+	pool := d.getOrCreatePool(atype)
+	if pool == nil {
+		return fmt.Errorf("no pipeline pool available for asset type %s", string(atype))
+	}
+
+	return pool.Dispatch(e)
+}
+
 func (d *dynamicDispatcher) getOrCreatePool(atype oam.AssetType) *pipelinePool {
 	d.RLock()
 	pool := d.pools[atype]
diff --git a/engine/plugins/api/dnsrepo.go b/engine/plugins/api/dnsrepo.go
@@ -1,4 +1,4 @@
-// Copyright © by Jeff Foley 2017-2025. All rights reserved.
+// Copyright © by Jeff Foley 2017-2026. All rights reserved.
 // Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
 // SPDX-License-Identifier: Apache-2.0
 
diff --git a/engine/plugins/horizontals/contact.go b/engine/plugins/horizontals/contact.go
@@ -76,13 +76,13 @@ func (h *horContact) lookup(e *et.Event, entity *dbt.Entity, since time.Time, co
 	var results []*scope.Association
 	if edges, err := e.Session.DB().OutgoingEdges(ctx, entity, since, labels...); err == nil && len(edges) > 0 {
 		for _, edge := range edges {
-			entity, err := e.Session.DB().FindEntityById(ctx, edge.ToEntity.ID)
+			to, err := e.Session.DB().FindEntityById(ctx, edge.ToEntity.ID)
 			if err != nil {
 				continue
 			}
 			// check if these asset discoveries could change the scope
 			if assocs, err := e.Session.Scope().IsAssociated(e.Session.DB(), &scope.Association{
-				Submission:  entity,
+				Submission:  to,
 				Confidence:  conf,
 				ScopeChange: true,
 			}); err == nil && len(assocs) > 0 {
diff --git a/engine/plugins/horizontals/fqdn.go b/engine/plugins/horizontals/fqdn.go
@@ -126,6 +126,7 @@ func (h *horfqdn) checkPTR(e *et.Event, edges []*dbt.Edge, fqdn *dbt.Entity, sin
 			}
 
 			for _, edge := range edges {
+				// acquire the FQDN asset that represents the IP address (a.k.a. the pointer record)
 				to, err := e.Session.DB().FindEntityById(ctx, edge.ToEntity.ID)
 				if err != nil {
 					continue
@@ -143,7 +144,7 @@ func (h *horfqdn) checkPTR(e *et.Event, edges []*dbt.Edge, fqdn *dbt.Entity, sin
 						if e.Session.Config().Active {
 							size = 250
 						}
-						h.plugin.submitIPAddresses(e, ip, h.plugin.source)
+						h.plugin.submitIPAddress(e, ip, h.plugin.source)
 						support.IPAddressSweep(e, ip, h.plugin.source, size, h.plugin.submitIPAddresses)
 						e.Session.Log().Info(fmt.Sprintf("[%s: %s] was added to the session scope", ip.AssetType(), ip.Key()))
 					}
diff --git a/engine/plugins/horizontals/plugin.go b/engine/plugins/horizontals/plugin.go
@@ -324,7 +324,7 @@ func (h *horizPlugin) sweepAroundIPs(ctx context.Context, e *et.Event, nb *dbt.E
 	}
 */
 
-func (h *horizPlugin) submitIPAddresses(e *et.Event, asset *oamnet.IPAddress, src *et.Source) {
+func (h *horizPlugin) submitIPAddress(e *et.Event, asset *oamnet.IPAddress, src *et.Source) {
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 
diff --git a/engine/plugins/known_fqdn.go b/engine/plugins/known_fqdn.go
@@ -1,4 +1,4 @@
-// Copyright © by Jeff Foley 2017-2025. All rights reserved.
+// Copyright © by Jeff Foley 2017-2026. All rights reserved.
 // Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
 // SPDX-License-Identifier: Apache-2.0
 
diff --git a/engine/plugins/support/support.go b/engine/plugins/support/support.go
@@ -284,6 +284,7 @@ func AddSLDInScope(e *et.Event) {
 		e.Meta = &FQDNMeta{
 			SLDInScope: true,
 		}
+		return
 	}
 
 	if fm, ok := e.Meta.(*FQDNMeta); ok {
diff --git a/engine/registry/pipelines.go b/engine/registry/pipelines.go
@@ -1,4 +1,4 @@
-// Copyright © by Jeff Foley 2017-2025. All rights reserved.
+// Copyright © by Jeff Foley 2017-2026. All rights reserved.
 // Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
 // SPDX-License-Identifier: Apache-2.0
 
diff --git a/engine/registry/registry.go b/engine/registry/registry.go
@@ -1,4 +1,4 @@
-// Copyright © by Jeff Foley 2017-2025. All rights reserved.
+// Copyright © by Jeff Foley 2017-2026. All rights reserved.
 // Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
 // SPDX-License-Identifier: Apache-2.0
 
diff --git a/engine/registry/registry_test.go b/engine/registry/registry_test.go
@@ -1,4 +1,4 @@
-// Copyright © by Jeff Foley 2017-2025. All rights reserved.
+// Copyright © by Jeff Foley 2017-2026. All rights reserved.
 // Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
 // SPDX-License-Identifier: Apache-2.0
 
diff --git a/engine/sessions/backlogdb/backlog.go b/engine/sessions/backlogdb/backlog.go
@@ -44,7 +44,7 @@ func (b *BacklogDB) Has(ctx context.Context, entityID string) (bool, error) {
 }
 
 // Enqueue inserts a new queued item if it does not already exist.
-// Idempotent: if entity_id already exists, it does nothing.
+// If entity_id already exists, it sets the state to queued.
 func (b *BacklogDB) Enqueue(ctx context.Context, atype oam.AssetType, entityID string) error {
 	if entityID == "" {
 		return ErrEmptyEntityID
@@ -53,8 +53,12 @@ func (b *BacklogDB) Enqueue(ctx context.Context, atype oam.AssetType, entityID s
 	_, err := b.db.ExecContext(ctx, `
 		INSERT INTO backlog_items(created_at, updated_at, etype, entity_id, state, lease_until)
 		VALUES(?, ?, ?, ?, ?, 0)
-		ON CONFLICT(entity_id) DO NOTHING`,
-		t, t, string(atype), entityID, StateQueued,
+		ON CONFLICT(entity_id) DO UPDATE SET
+			updated_at  = excluded.updated_at,
+			state       = ?,
+			lease_owner = NULL,
+			lease_until = 0`,
+		t, t, string(atype), entityID, StateQueued, StateQueued,
 	)
 	return err
 }
@@ -69,10 +73,10 @@ func (b *BacklogDB) EnqueueDone(ctx context.Context, atype oam.AssetType, entity
 		INSERT INTO backlog_items(created_at, updated_at, etype, entity_id, state, lease_until)
 		VALUES(?, ?, ?, ?, ?, 0)
 		ON CONFLICT(entity_id) DO UPDATE SET
-			updated_at = excluded.updated_at,
-			state      = ?,
-			lease_owner= NULL,
-			lease_until= 0`,
+			updated_at  = excluded.updated_at,
+			state       = ?,
+			lease_owner = NULL,
+			lease_until = 0`,
 		t, t, string(atype), entityID, StateDone, StateDone,
 	)
 	return err
diff --git a/engine/sessions/scope/assoc.go b/engine/sessions/scope/assoc.go
@@ -312,7 +312,7 @@ func (s *Scope) awayFromAssetsWithAssociation(c repository.Repository, assoc *db
 		}
 	}
 	if len(results) == 0 {
-		return nil, errors.New("zero assets were found in-scope one hop forward from the provided asset")
+		return nil, errors.New("zero assets were found in-scope one hop away from the provided asset")
 	}
 	return results, nil
 }
@@ -375,7 +375,7 @@ func (s *Scope) towardsAssetsWithAssociation(c repository.Repository, asset *dbt
 		}
 	}
 	if len(results) == 0 {
-		return nil, errors.New("zero assets were found in-scope one hop forward from the provided asset")
+		return nil, errors.New("zero assets were found in-scope one hop toward the provided asset")
 	}
 	return results, nil
 }
diff --git a/engine/sessions/scope/locations.go b/engine/sessions/scope/locations.go
@@ -45,16 +45,17 @@ func (s *Scope) Locations() []*contact.Location {
 }
 
 func (s *Scope) matchesLocation(loc *contact.Location, conf int) (oam.Asset, int) {
+	if loc.BuildingNumber == "" {
+		return nil, 0
+	}
+
 	for _, loc2 := range s.Locations() {
-		if loc.BuildingNumber == "" || loc2.BuildingNumber == "" || loc.BuildingNumber != loc2.BuildingNumber {
+		if loc.BuildingNumber != loc2.BuildingNumber {
 			continue
 		}
 
-		lstr1 := fmt.Sprintf("%s %s %s %s %s", loc.BuildingNumber,
-			loc.StreetName, loc.City, loc.Province, loc.PostalCode)
-		lstr2 := fmt.Sprintf("%s %s %s %s %s", loc2.BuildingNumber,
-			loc2.StreetName, loc2.City, loc2.Province, loc2.PostalCode)
-
+		lstr1 := fmt.Sprintf("%s %s %s %s", loc.BuildingNumber, loc.StreetName, loc.City, loc.Province)
+		lstr2 := fmt.Sprintf("%s %s %s %s", loc2.BuildingNumber, loc2.StreetName, loc2.City, loc2.Province)
 		if strings.EqualFold(lstr1, lstr2) {
 			return loc2, 100
 		}

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-// Copyright © by Jeff Foley 2017-2025. All rights reserved.`
	`1`	`+// Copyright © by Jeff Foley 2017-2026. All rights reserved.`
`2`	`2`	`// Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.`
`3`	`3`	`// SPDX-License-Identifier: Apache-2.0`
`4`	`4`
Original file line number	Diff line number	Diff line change
`@@ -126,6 +126,7 @@ func (h horfqdn) checkPTR(e et.Event, edges []dbt.Edge, fqdn dbt.Entity, sin`
`126`	`126`	`}`
`127`	`127`
`128`	`128`	`for _, edge := range edges {`
	`129`	`+ // acquire the FQDN asset that represents the IP address (a.k.a. the pointer record)`
`129`	`130`	`to, err := e.Session.DB().FindEntityById(ctx, edge.ToEntity.ID)`
`130`	`131`	`if err != nil {`
`131`	`132`	`continue`
`@@ -143,7 +144,7 @@ func (h horfqdn) checkPTR(e et.Event, edges []dbt.Edge, fqdn dbt.Entity, sin`
`143`	`144`	`if e.Session.Config().Active {`
`144`	`145`	`size = 250`
`145`	`146`	`}`
`146`		`- h.plugin.submitIPAddresses(e, ip, h.plugin.source)`
	`147`	`+ h.plugin.submitIPAddress(e, ip, h.plugin.source)`
`147`	`148`	`support.IPAddressSweep(e, ip, h.plugin.source, size, h.plugin.submitIPAddresses)`
`148`	`149`	`e.Session.Log().Info(fmt.Sprintf("[%s: %s] was added to the session scope", ip.AssetType(), ip.Key()))`
`149`	`150`	`}`
Original file line number	Diff line number	Diff line change
`@@ -324,7 +324,7 @@ func (h horizPlugin) sweepAroundIPs(ctx context.Context, e et.Event, nb *dbt.E`
`324`	`324`	`}`
`325`	`325`	`*/`
`326`	`326`
`327`		`-func (h horizPlugin) submitIPAddresses(e et.Event, asset oamnet.IPAddress, src et.Source) {`
	`327`	`+func (h horizPlugin) submitIPAddress(e et.Event, asset oamnet.IPAddress, src et.Source) {`
`328`	`328`	`ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)`
`329`	`329`	`defer cancel()`
`330`	`330`
Original file line number	Diff line number	Diff line change
`@@ -284,6 +284,7 @@ func AddSLDInScope(e *et.Event) {`
`284`	`284`	`e.Meta = &FQDNMeta{`
`285`	`285`	`SLDInScope: true,`
`286`	`286`	`}`
	`287`	`+ return`
`287`	`288`	`}`
`288`	`289`
`289`	`290`	`if fm, ok := e.Meta.(*FQDNMeta); ok {`