add mtx to protect concurrent access

schema.go

@@ -72,6 +72,7 @@ func (s *Schema) Register(uri string, registry *SchemaRegistry) {
 	if uri != "" && address != "" {
 		address, _ = SafeResolveURL(uri, address)
 	}
+
 	if s.docPath == "" && address != "" && address[0] != '#' {
 		docURI := ""
 		if u, err := url.Parse(address); err != nil {
@@ -326,6 +327,7 @@ func (s *Schema) ValidateBytes(ctx context.Context, data []byte) ([]KeyError, er
 	if err := json.Unmarshal(data, &doc); err != nil {
 		return nil, fmt.Errorf("error parsing JSON bytes: %w", err)
 	}
+
 	vs := s.Validate(ctx, doc)
 	return *vs.Errs, nil
 }

schema_registry.go

@@ -4,37 +4,42 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"sync"
 )
 
 var sr *SchemaRegistry
 
+func init() {
+	// sets the global sr variable the first time the module is used;
+	// this fixes a concurrency issue.
+	sr = &SchemaRegistry{
+		mtx:           sync.RWMutex{},
+		schemaLookup:  map[string]*Schema{},
+		contextLookup: map[string]*Schema{},
+	}
+}
+
 // SchemaRegistry maintains a lookup table between schema string references
 // and actual schemas
 type SchemaRegistry struct {
+	mtx           sync.RWMutex
 	schemaLookup  map[string]*Schema
 	contextLookup map[string]*Schema
 }
 
 // GetSchemaRegistry provides an accessor to a globally available schema registry
 func GetSchemaRegistry() *SchemaRegistry {
-	if sr == nil {
-		sr = &SchemaRegistry{
-			schemaLookup:  map[string]*Schema{},
-			contextLookup: map[string]*Schema{},
-		}
-	}
 	return sr
 }
 
-// ResetSchemaRegistry resets the main SchemaRegistry
-func ResetSchemaRegistry() {
-	sr = nil
-}
-
 // Get fetches a schema from the top level context registry or fetches it from a remote
 func (sr *SchemaRegistry) Get(ctx context.Context, uri string) *Schema {
 	uri = strings.TrimRight(uri, "#")
+
+	sr.mtx.RLock()
 	schema := sr.schemaLookup[uri]
+	sr.mtx.RUnlock()
+
 	if schema == nil {
 		fetchedSchema := &Schema{}
 		err := FetchSchema(ctx, uri, fetchedSchema)
@@ -48,20 +53,31 @@ func (sr *SchemaRegistry) Get(ctx context.Context, uri string) *Schema {
 		fetchedSchema.docPath = uri
 		// TODO(arqu): meta validate schema
 		schema = fetchedSchema
+
+		sr.mtx.Lock()
 		sr.schemaLookup[uri] = schema
+		sr.mtx.Unlock()
 	}
 	return schema
 }
 
 // GetKnown fetches a schema from the top level context registry
 func (sr *SchemaRegistry) GetKnown(uri string) *Schema {
 	uri = strings.TrimRight(uri, "#")
+
+	sr.mtx.RLock()
+	defer sr.mtx.RUnlock()
+
 	return sr.schemaLookup[uri]
 }
 
 // GetLocal fetches a schema from the local context registry
 func (sr *SchemaRegistry) GetLocal(uri string) *Schema {
 	uri = strings.TrimRight(uri, "#")
+
+	sr.mtx.RLock()
+	defer sr.mtx.RUnlock()
+
 	return sr.contextLookup[uri]
 }
 
@@ -70,16 +86,24 @@ func (sr *SchemaRegistry) Register(sch *Schema) {
 	if sch.docPath == "" {
 		return
 	}
+
+	sr.mtx.Lock()
 	sr.schemaLookup[sch.docPath] = sch
+	sr.mtx.Unlock()
 }
 
 // RegisterLocal registers a schema to a local context
 func (sr *SchemaRegistry) RegisterLocal(sch *Schema) {
 	if sch.id != "" && IsLocalSchemaID(sch.id) {
+		sr.mtx.Lock()
 		sr.contextLookup[sch.id] = sch
+		sr.mtx.Unlock()
 	}
 
 	if sch.HasKeyword("$anchor") {
+		sr.mtx.Lock()
+		defer sr.mtx.Unlock()
+
 		anchorKeyword := sch.keywords["$anchor"].(*Anchor)
 		anchorURI := sch.docPath + "#" + string(*anchorKeyword)
 		if sr.contextLookup == nil {