fix: panic with concurrent schema parsing (#502)

hamba · Feb 12, 2025 · b48c057 · b48c057
1 parent 323f149
commit b48c057
Show file tree

Hide file tree

Showing 4 changed files with 155 additions and 1 deletion.
diff --git a/schema.go b/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
 

diff --git a/schema_parse.go b/schema_parse.go
@@ -74,11 +74,17 @@ func ParseBytesWithCache(schema []byte, namespace string, cache *SchemaCache) (S
 		json = string(schema)
 	}
 
+	internalCache := &SchemaCache{}
+	internalCache.AddAll(cache)
+
 	seen := seenCache{}
-	s, err := parseType(namespace, json, seen, cache)
+	s, err := parseType(namespace, json, seen, internalCache)
 	if err != nil {
 		return nil, err
 	}
+
+	cache.AddAll(internalCache)
+
 	return derefSchema(s), nil
 }
 

diff --git a/schema_test.go b/schema_test.go
@@ -3,6 +3,7 @@ package avro_test
 import (
 	"encoding/json"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/hamba/avro/v2"
@@ -2244,3 +2245,18 @@ func TestNewSchema_IgnoresInvalidProperties(t *testing.T) {
 		}, rec.Props())
 	})
 }
+
+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.ParseFiles("testdata/concurrent-schema.avsc")
+			require.NoError(t, err)
+		}()
+	}
+
+	wg.Wait()
+}
diff --git a/testdata/concurrent-schema.avsc b/testdata/concurrent-schema.avsc
@@ -0,0 +1,121 @@
+{
+  "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": {}
+    }
+  ]
+}