bugfix: Connect() panics on schema update

Part of #278

Author: oleg-jukovec

CHANGELOG.md

@@ -24,6 +24,7 @@ Versioning](http://semver.org/spec/v2.0.0.html) except to the first release.
 - ConnectionPool does not properly handle disconnection with Opts.Reconnect
   set (#272)
 - Watcher events loss with a small per-request timeout (#284)
+- Connect() panics on concurrent schema update (#278)
 
 ## [1.10.0] - 2022-12-31
 

config.lua

@@ -181,6 +181,23 @@ local function push_func(cnt)
 end
 rawset(_G, 'push_func', push_func)
 
+local function create_spaces()
+    for i=1,10 do
+        local s = box.schema.space.create('test' .. tostring(i), {
+            id = 700 + i,
+            if_not_exists = true,
+        })
+        local idx = s:create_index('test' .. tostring(i) .. 'primary', {
+            type = 'tree',
+            parts = {1, 'uint'},
+            if_not_exists = true
+        })
+        idx:drop()
+        s:drop()
+    end
+end
+rawset(_G, 'create_spaces', create_spaces)
+
 local function tarantool_version_at_least(wanted_major, wanted_minor, wanted_patch)
     -- https://github.com/tarantool/crud/blob/733528be02c1ffa3dacc12c034ee58c9903127fc/test/helper.lua#L316-L337
     local major_minor_patch = _TARANTOOL:split('-', 1)[1]

schema.go

@@ -325,8 +325,13 @@ func (conn *Connection) loadSchema() (err error) {
 		return err
 	}
 	for _, index := range indexes {
-		schema.SpacesById[index.SpaceId].IndexesById[index.Id] = index
-		schema.SpacesById[index.SpaceId].Indexes[index.Name] = index
+		spaceId := index.SpaceId
+		if _, ok := schema.SpacesById[spaceId]; ok {
+			schema.SpacesById[spaceId].IndexesById[index.Id] = index
+			schema.SpacesById[spaceId].Indexes[index.Name] = index
+		} else {
+			return errors.New("concurrent schema update")
+		}
 	}
 
 	conn.lockShards()

tarantool_test.go

@@ -3894,6 +3894,29 @@ func TestWatcher_Unregister_concurrent(t *testing.T) {
 	wg.Wait()
 }
 
+func TestConnect_schema_update(t *testing.T) {
+	conn := test_helpers.ConnectWithValidation(t, server, opts)
+	defer conn.Close()
+
+	for i := 0; i < 100; i++ {
+		fut := conn.Do(NewCallRequest("create_spaces"))
+
+		if conn, err := Connect(server, opts); err != nil {
+			if err.Error() != "concurrent schema update" {
+				t.Errorf("unexpected error: %s", err)
+			}
+		} else if conn == nil {
+			t.Errorf("conn is nil")
+		} else {
+			conn.Close()
+		}
+
+		if _, err := fut.Get(); err != nil {
+			t.Errorf("Failed to call create_spaces: %s", err)
+		}
+	}
+}
+
 // runTestMain is a body of TestMain function
 // (see https://pkg.go.dev/testing#hdr-Main).
 // Using defer + os.Exit is not works so TestMain body