fix: panic with concurrent schema parsing

concurrent_parse_test.go

@@ -0,0 +1,148 @@
+package avro_test
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/hamba/avro/v2"
+	"github.com/stretchr/testify/require"
+)
+
+func TestConcurrentParse(t *testing.T) {
+	var wg sync.WaitGroup
+
+	for i := 0; i < 10000; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			_, err := avro.Parse(testSchema())
+			require.NoError(t, err)
+		}()
+	}
+
+	wg.Wait()
+}
+
+func testSchema() string {
+	return `{
+  "type": "record",
+  "name": "FootballUpdateEvent",
+  "namespace": "com.example.avro",
+  "fields": [
+    {
+      "name": "event_metadata",
+      "type": {
+        "type": "record",
+        "name": "EventMetadata",
+        "fields": [
+          { "name": "name", "type": "string", "default": "" },
+          { "name": "trace", "type": "string", "default": "" },
+          { "name": "stamp", "type": "long", "default": 0 },
+          { "name": "destination", "type": "string", "default": "" }
+        ]
+      },
+      "default": {
+        "name": "",
+        "trace": "",
+        "stamp": 0,
+        "destination": ""
+      }
+    },
+    {
+      "name": "player",
+      "type": {
+        "type": "record",
+        "name": "Player",
+        "fields": [
+          { "name": "team_id", "type": "string", "default": "" },
+          { "name": "name", "type": "string", "default": "" },
+          { "name": "team", "type": "string", "default": "" },
+          { "name": "contract", "type": "long", "default": 0 },
+          { "name": "xg", "type": "double", "default": 0.0 },
+          { "name": "xgp", "type": "double", "default": 0.0 },
+          { "name": "xgp99", "type": "double", "default": 0.0 },
+          { "name": "xg90", "type": "double", "default": 0.0 },
+          {
+            "name": "xg90p",
+            "type": {
+              "type": "record",
+              "name": "MatchXG",
+              "fields": [
+                { "name": "matchXG", "type": "double", "default": 0.0 },
+                { "name": "matchXGP", "type": "string", "default": "" }
+              ]
+            },
+            "default": {
+              "matchXG": 0.0,
+              "matchXGP": ""
+            }
+          },
+          {
+            "name": "ttd",
+            "type": "MatchXG",
+            "default": {
+              "matchXG": 0.0,
+              "matchXGP": ""
+            }
+          },
+          {
+            "name": "leagueXG",
+            "type": {
+              "type": "record",
+              "name": "LeagueXG",
+              "fields": [
+                { "name": "top_assist", "type": "string", "default": "" },
+                { "name": "top_score", "type": "string", "default": "" },
+                { "name": "top_xg", "type": "string", "default": "" },
+                { "name": "top_creation", "type": "string", "default": "" }
+              ]
+            },
+            "default": {
+              "top_assist": "",
+              "top_score": "",
+              "top_xg": "",
+              "top_creation": ""
+            }
+          },
+          {
+            "name": "player_numbers",
+            "type": {
+              "type": "array",
+              "items": {
+                "type": "record",
+                "name": "PlayerNumber",
+                "fields": [
+                  { "name": "player_number", "type": "long", "default": 0 }
+                ]
+              }
+            },
+            "default": []
+          },
+          {
+            "name": "contact_renewal",
+            "type": [
+              "null",
+              {
+                "type": "record",
+                "name": "ContactRenewal",
+                "fields": [
+                  { "name": "stamp", "type": "long", "default": 0 },
+                  { "name": "type", "type": "string", "default": "" },
+                  { "name": "xg", "type": "MatchXG", "default": {
+                    "matchXG": 0.0,
+                    "matchXGP": ""
+                  } }
+                ]
+              }
+            ],
+            "default": null
+          },
+          { "name": "defence", "type": "string", "default": "" },
+          { "name": "offence", "type": "string", "default": "" }
+        ]
+      },
+      "default": {}
+    }
+  ]
+}`
+}

schema.go

@@ -131,6 +131,17 @@ func (c *SchemaCache) Get(name string) Schema {
 	return nil
 }
 
+// AddAll adds all schemas from the given cache to the current cache.
+func (c *SchemaCache) AddAll(cache *SchemaCache) {
+	if cache == nil {
+		return
+	}
+	cache.cache.Range(func(key, value interface{}) bool {
+		c.cache.Store(key, value)
+		return true
+	})
+}
+
 // Schemas is a slice of Schemas.
 type Schemas []Schema
 

schema_parse.go

@@ -29,7 +29,7 @@ func Parse(schema string) (Schema, error) {
 
 // ParseWithCache parses a schema string using the given namespace and schema cache.
 func ParseWithCache(schema, namespace string, cache *SchemaCache) (Schema, error) {
-	return ParseBytesWithCache([]byte(schema), namespace, cache)
+	return ParseBytesWithCache([]byte(schema), namespace)
 }
 
 // MustParse parses a schema string, panicking if there is an error.
@@ -64,21 +64,27 @@ func ParseFiles(paths ...string) (Schema, error) {
 
 // ParseBytes parses a schema byte slice.
 func ParseBytes(schema []byte) (Schema, error) {
-	return ParseBytesWithCache(schema, "", DefaultSchemaCache)
+	return ParseBytesWithCache(schema, "")
 }
 
 // ParseBytesWithCache parses a schema byte slice using the given namespace and schema cache.
-func ParseBytesWithCache(schema []byte, namespace string, cache *SchemaCache) (Schema, error) {
+func ParseBytesWithCache(schema []byte, namespace string) (Schema, error) {
 	var json any
 	if err := jsoniter.Unmarshal(schema, &json); err != nil {
 		json = string(schema)
 	}
 
+	internalCache := &SchemaCache{}
+	internalCache.AddAll(DefaultSchemaCache)
+
 	seen := seenCache{}
-	s, err := parseType(namespace, json, seen, cache)
+	s, err := parseType(namespace, json, seen, internalCache)
 	if err != nil {
 		return nil, err
 	}
+
+	DefaultSchemaCache.AddAll(internalCache)
+
 	return derefSchema(s), nil
 }