From 19f75bb6cab13520826f47b432f8a274b84238aa Mon Sep 17 00:00:00 2001
From: Patrick Pacher <patrick.pacher@gmail.com>
Date: Fri, 18 Sep 2020 12:11:05 +0200
Subject: [PATCH] Fix locking in runtime and hashmap storage

---
 database/storage/hashmap/map.go | 15 +++++++--------
 runtime/registry.go             | 21 +++++++--------------
 2 files changed, 14 insertions(+), 22 deletions(-)

diff --git a/database/storage/hashmap/map.go b/database/storage/hashmap/map.go
index fb1e073..96d8390 100644
--- a/database/storage/hashmap/map.go
+++ b/database/storage/hashmap/map.go
@@ -104,17 +104,16 @@ func (hm *HashMap) queryExecutor(queryIter *iterator.Iterator, q *query.Query, l
 
 mapLoop:
 	for key, record := range hm.db {
+		record.Lock()
+		if !q.MatchesKey(key) ||
+			!q.MatchesRecord(record) ||
+			!record.Meta().CheckValidity() ||
+			!record.Meta().CheckPermission(local, internal) {
 
-		switch {
-		case !q.MatchesKey(key):
-			continue
-		case !q.MatchesRecord(record):
-			continue
-		case !record.Meta().CheckValidity():
-			continue
-		case !record.Meta().CheckPermission(local, internal):
+			record.Unlock()
 			continue
 		}
+		record.Unlock()
 
 		select {
 		case <-queryIter.Done:
diff --git a/runtime/registry.go b/runtime/registry.go
index 21f8a02..445185a 100644
--- a/runtime/registry.go
+++ b/runtime/registry.go
@@ -202,23 +202,16 @@ func (r *Registry) Query(q *query.Query, local, internal bool) (*iterator.Iterat
 			}
 
 			for _, r := range records {
-				// TODO(ppacher): do we need to lock r?
-				//                storage/hashmap does not lock the records
-				//                before sending them to the iterator but
-				//                better make sure that's correct.
+				r.Lock()
+				if q.MatchesKey(r.DatabaseKey()) ||
+					!r.Meta().CheckValidity() ||
+					!r.Meta().CheckPermission(local, internal) ||
+					!q.MatchesRecord(r) {
 
-				if !strings.HasPrefix(r.DatabaseKey(), searchPrefix) {
-					continue
-				}
-				if !r.Meta().CheckValidity() {
-					continue
-				}
-				if !r.Meta().CheckPermission(local, internal) {
-					continue
-				}
-				if !q.MatchesRecord(r) {
+					r.Unlock()
 					continue
 				}
+				r.Unlock()
 
 				select {
 				case iter.Next <- r: