Skip to content

Commit a5bfe5d

Browse files
peterbroadhurstpeterbroadhurst
committed
Re-order listener creation logic to do concurrency protection in core
Signed-off-by: Peter Broadhurst <peter.broadhurst@kaleido.io>
1 parent 28b672f commit a5bfe5d

2 files changed

Lines changed: 205 additions & 5 deletions

File tree

‎internal/contracts/manager.go‎

Lines changed: 23 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1106,13 +1106,32 @@ func (cm *contractManager) AddContractListener(ctx context.Context, listener *co
11061106
return nil, err
11071107
}
11081108

1109-
if err = cm.blockchain.AddContractListener(ctx, &listener.ContractListener, ""); err != nil {
1110-
return nil, err
1111-
}
11121109
if listener.Name == "" {
1113-
listener.Name = listener.BackendID
1110+
listener.Name = listener.ID.String() // default to the same name as the ID (cannot use the backendID here as not created yet)
11141111
}
1112+
1113+
// Placeholder backend ID until we've created in the backend - which happens after we store in the DB (see below)
1114+
listener.BackendID = "pending-" + listener.ID.String()
1115+
1116+
// Create the listener in our DB, checking for duplicates (only one concurrent API can win here)
11151117
if err = cm.database.InsertContractListener(ctx, verifiedContractListener); err != nil {
1118+
if existing, lookupErr := cm.database.GetContractListener(ctx, cm.namespace, listener.Name); lookupErr == nil && existing != nil && !existing.ID.Equals(listener.ID) {
1119+
return nil, i18n.NewError(ctx, coremsgs.MsgContractListenerNameExists, cm.namespace, listener.Name)
1120+
}
1121+
return nil, err
1122+
}
1123+
1124+
if err = cm.blockchain.AddContractListener(ctx, &listener.ContractListener, ""); err != nil {
1125+
// Cleanup our DB record before return, in case of a failure to create in the backend
1126+
if delErr := cm.database.DeleteContractListenerByID(ctx, cm.namespace, listener.ID); delErr != nil {
1127+
log.L(ctx).Errorf("Failed to delete listener %s after connector subscription creation failed: %s", listener.ID, delErr)
1128+
}
1129+
return nil, err
1130+
}
1131+
1132+
// Store the backend ID now we've created it
1133+
if err = cm.database.UpdateContractListener(ctx, cm.namespace, listener.ID,
1134+
database.ContractListenerQueryFactory.NewUpdate(ctx).Set("backendid", listener.BackendID)); err != nil {
11161135
return nil, err
11171136
}
11181137

‎internal/contracts/manager_test.go‎

Lines changed: 182 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -806,6 +806,7 @@ func TestAddContractListenerInline(t *testing.T) {
806806
mdi.On("GetContractListeners", context.Background(), "ns1", mock.Anything).Return(nil, nil, nil)
807807
mbi.On("AddContractListener", context.Background(), &sub.ContractListener, "").Return(nil)
808808
mdi.On("InsertContractListener", context.Background(), &sub.ContractListener).Return(nil)
809+
mdi.On("UpdateContractListener", context.Background(), "ns1", mock.Anything, mock.Anything).Return(nil)
809810

810811
result, err := cm.AddContractListener(context.Background(), sub)
811812
assert.NoError(t, err)
@@ -847,6 +848,7 @@ func TestAddContractListenerInlineNilLocation(t *testing.T) {
847848
return cl.Location == nil
848849
}), "").Return(nil)
849850
mdi.On("InsertContractListener", context.Background(), &sub.ContractListener).Return(nil)
851+
mdi.On("UpdateContractListener", context.Background(), "ns1", mock.Anything, mock.Anything).Return(nil)
850852

851853
result, err := cm.AddContractListener(context.Background(), sub)
852854
assert.NoError(t, err)
@@ -885,6 +887,7 @@ func TestAddContractListenerNoLocationOK(t *testing.T) {
885887
mdi.On("GetContractListeners", context.Background(), "ns1", mock.Anything).Return(nil, nil, nil)
886888
mbi.On("AddContractListener", context.Background(), &sub.ContractListener, "").Return(nil)
887889
mdi.On("InsertContractListener", context.Background(), &sub.ContractListener).Return(nil)
890+
mdi.On("UpdateContractListener", context.Background(), "ns1", mock.Anything, mock.Anything).Return(nil)
888891

889892
result, err := cm.AddContractListener(context.Background(), sub)
890893
assert.NoError(t, err)
@@ -937,6 +940,7 @@ func TestAddContractListenerByEventPath(t *testing.T) {
937940
mdi.On("GetFFIByID", context.Background(), "ns1", interfaceID).Return(&fftypes.FFI{}, nil)
938941
mdi.On("GetFFIEvent", context.Background(), "ns1", interfaceID, sub.EventPath).Return(event, nil)
939942
mdi.On("InsertContractListener", context.Background(), &sub.ContractListener).Return(nil)
943+
mdi.On("UpdateContractListener", context.Background(), "ns1", mock.Anything, mock.Anything).Return(nil)
940944

941945
result, err := cm.AddContractListener(context.Background(), sub)
942946
assert.NoError(t, err)
@@ -1648,7 +1652,10 @@ func TestAddContractListenerBlockchainFail(t *testing.T) {
16481652
mbi.On("GenerateEventSignature", context.Background(), mock.Anything).Return("changed", nil)
16491653
mbi.On("GenerateEventSignatureWithLocation", context.Background(), mock.Anything, mock.Anything).Return("0x123:changed", nil)
16501654
mdi.On("GetContractListeners", context.Background(), "ns1", mock.Anything).Return(nil, nil, nil)
1655+
mdi.On("InsertContractListener", context.Background(), &sub.ContractListener).Return(nil)
16511656
mbi.On("AddContractListener", context.Background(), &sub.ContractListener, "").Return(fmt.Errorf("pop"))
1657+
// The connector failure compensates by deleting the row inserted above
1658+
mdi.On("DeleteContractListenerByID", context.Background(), "ns1", mock.Anything).Return(nil)
16521659

16531660
_, err := cm.AddContractListener(context.Background(), sub)
16541661
assert.EqualError(t, err, "pop")
@@ -1686,8 +1693,181 @@ func TestAddContractListenerUpsertSubFail(t *testing.T) {
16861693
mbi.On("GenerateEventSignature", context.Background(), mock.Anything).Return("changed", nil)
16871694
mbi.On("GenerateEventSignatureWithLocation", context.Background(), mock.Anything, mock.Anything).Return("0x123:changed", nil)
16881695
mdi.On("GetContractListeners", context.Background(), "ns1", mock.Anything).Return(nil, nil, nil)
1689-
mbi.On("AddContractListener", context.Background(), &sub.ContractListener, "").Return(nil)
16901696
mdi.On("InsertContractListener", context.Background(), &sub.ContractListener).Return(fmt.Errorf("pop"))
1697+
mdi.On("GetContractListener", context.Background(), "ns1", mock.Anything).Return(nil, nil)
1698+
1699+
_, err := cm.AddContractListener(context.Background(), sub)
1700+
assert.EqualError(t, err, "pop")
1701+
1702+
mbi.AssertExpectations(t)
1703+
mdi.AssertExpectations(t)
1704+
}
1705+
1706+
func TestAddContractListenerInsertNameRace(t *testing.T) {
1707+
cm := newTestContractManager()
1708+
mbi := cm.blockchain.(*blockchainmocks.Plugin)
1709+
mdi := cm.database.(*databasemocks.Plugin)
1710+
1711+
sub := &core.ContractListenerInput{
1712+
ContractListener: core.ContractListener{
1713+
Name: "sub1",
1714+
Location: fftypes.JSONAnyPtr(fftypes.JSONObject{
1715+
"address": "0x123",
1716+
}.String()),
1717+
Event: &core.FFISerializedEvent{
1718+
FFIEventDefinition: fftypes.FFIEventDefinition{
1719+
Name: "changed",
1720+
Params: fftypes.FFIParams{
1721+
{
1722+
Name: "value",
1723+
Schema: fftypes.JSONAnyPtr(`{"type": "integer"}`),
1724+
},
1725+
},
1726+
},
1727+
},
1728+
Topic: "test-topic",
1729+
},
1730+
}
1731+
1732+
mbi.On("NormalizeContractLocation", context.Background(), blockchain.NormalizeListener, sub.Location).Return(sub.Location, nil)
1733+
mbi.On("GenerateEventSignature", context.Background(), mock.Anything).Return("changed", nil)
1734+
mbi.On("GenerateEventSignatureWithLocation", context.Background(), mock.Anything, mock.Anything).Return("0x123:changed", nil)
1735+
mdi.On("GetContractListeners", context.Background(), "ns1", mock.Anything).Return(nil, nil, nil)
1736+
mdi.On("GetContractListener", context.Background(), "ns1", "sub1").Return(nil, nil).Once()
1737+
mdi.On("InsertContractListener", context.Background(), &sub.ContractListener).Return(fmt.Errorf("unique constraint violation")) // sim another thread winning
1738+
mdi.On("GetContractListener", context.Background(), "ns1", "sub1").Return(&core.ContractListener{ID: fftypes.NewUUID(), Name: "sub1"}, nil).Once()
1739+
1740+
_, err := cm.AddContractListener(context.Background(), sub)
1741+
assert.Regexp(t, "FF10312", err)
1742+
1743+
mbi.AssertExpectations(t)
1744+
mdi.AssertExpectations(t)
1745+
}
1746+
1747+
func TestAddContractListenerBlockchainFailDeleteFail(t *testing.T) {
1748+
cm := newTestContractManager()
1749+
mbi := cm.blockchain.(*blockchainmocks.Plugin)
1750+
mdi := cm.database.(*databasemocks.Plugin)
1751+
1752+
sub := &core.ContractListenerInput{
1753+
ContractListener: core.ContractListener{
1754+
Location: fftypes.JSONAnyPtr(fftypes.JSONObject{
1755+
"address": "0x123",
1756+
}.String()),
1757+
Event: &core.FFISerializedEvent{
1758+
FFIEventDefinition: fftypes.FFIEventDefinition{
1759+
Name: "changed",
1760+
Params: fftypes.FFIParams{
1761+
{
1762+
Name: "value",
1763+
Schema: fftypes.JSONAnyPtr(`{"type": "integer"}`),
1764+
},
1765+
},
1766+
},
1767+
},
1768+
Topic: "test-topic",
1769+
},
1770+
}
1771+
1772+
mbi.On("NormalizeContractLocation", context.Background(), blockchain.NormalizeListener, sub.Location).Return(sub.Location, nil)
1773+
mbi.On("GenerateEventSignature", context.Background(), mock.Anything).Return("changed", nil)
1774+
mbi.On("GenerateEventSignatureWithLocation", context.Background(), mock.Anything, mock.Anything).Return("0x123:changed", nil)
1775+
mdi.On("GetContractListeners", context.Background(), "ns1", mock.Anything).Return(nil, nil, nil)
1776+
mdi.On("InsertContractListener", context.Background(), &sub.ContractListener).Return(nil)
1777+
mbi.On("AddContractListener", context.Background(), &sub.ContractListener, "").Return(fmt.Errorf("pop")) // backend failure
1778+
mdi.On("DeleteContractListenerByID", context.Background(), "ns1", mock.Anything).Return(fmt.Errorf("delete failed")) // deletion failure
1779+
1780+
_, err := cm.AddContractListener(context.Background(), sub)
1781+
assert.EqualError(t, err, "pop")
1782+
1783+
mbi.AssertExpectations(t)
1784+
mdi.AssertExpectations(t)
1785+
}
1786+
1787+
func TestAddContractListenerUnnamedProvisionalBackendID(t *testing.T) {
1788+
cm := newTestContractManager()
1789+
mbi := cm.blockchain.(*blockchainmocks.Plugin)
1790+
mdi := cm.database.(*databasemocks.Plugin)
1791+
1792+
sub := &core.ContractListenerInput{
1793+
ContractListener: core.ContractListener{
1794+
Location: fftypes.JSONAnyPtr(fftypes.JSONObject{
1795+
"address": "0x123",
1796+
}.String()),
1797+
Event: &core.FFISerializedEvent{
1798+
FFIEventDefinition: fftypes.FFIEventDefinition{
1799+
Name: "changed",
1800+
Params: fftypes.FFIParams{
1801+
{
1802+
Name: "value",
1803+
Schema: fftypes.JSONAnyPtr(`{"type": "integer"}`),
1804+
},
1805+
},
1806+
},
1807+
},
1808+
Topic: "test-topic",
1809+
},
1810+
}
1811+
1812+
mbi.On("NormalizeContractLocation", context.Background(), blockchain.NormalizeListener, sub.Location).Return(sub.Location, nil)
1813+
mbi.On("GenerateEventSignature", context.Background(), mock.Anything).Return("changed", nil)
1814+
mbi.On("GenerateEventSignatureWithLocation", context.Background(), mock.Anything, mock.Anything).Return("0x123:changed", nil)
1815+
mdi.On("GetContractListeners", context.Background(), "ns1", mock.Anything).Return(nil, nil, nil)
1816+
// The unnamed listener is inserted with its ID as the name, and a unique provisional backend ID
1817+
mdi.On("InsertContractListener", context.Background(), mock.MatchedBy(func(l *core.ContractListener) bool {
1818+
return l.Name == l.ID.String() && l.BackendID == "pending-"+l.ID.String()
1819+
})).Return(nil)
1820+
// The connector assigns the real backend ID, which is written back to the row
1821+
mbi.On("AddContractListener", context.Background(), mock.MatchedBy(func(l *core.ContractListener) bool {
1822+
l.BackendID = "sb-12345"
1823+
return true
1824+
}), "").Return(nil)
1825+
mdi.On("UpdateContractListener", context.Background(), "ns1", mock.Anything, mock.MatchedBy(func(u ffapi.Update) bool {
1826+
uu, _ := u.Finalize()
1827+
return strings.Contains(uu.String(), "sb-12345")
1828+
})).Return(nil)
1829+
1830+
result, err := cm.AddContractListener(context.Background(), sub)
1831+
assert.NoError(t, err)
1832+
assert.Equal(t, result.ID.String(), result.Name)
1833+
assert.Equal(t, "sb-12345", result.BackendID)
1834+
1835+
mbi.AssertExpectations(t)
1836+
mdi.AssertExpectations(t)
1837+
}
1838+
1839+
func TestAddContractListenerUpdateBackendIDFail(t *testing.T) {
1840+
cm := newTestContractManager()
1841+
mbi := cm.blockchain.(*blockchainmocks.Plugin)
1842+
mdi := cm.database.(*databasemocks.Plugin)
1843+
1844+
sub := &core.ContractListenerInput{
1845+
ContractListener: core.ContractListener{
1846+
Location: fftypes.JSONAnyPtr(fftypes.JSONObject{
1847+
"address": "0x123",
1848+
}.String()),
1849+
Event: &core.FFISerializedEvent{
1850+
FFIEventDefinition: fftypes.FFIEventDefinition{
1851+
Name: "changed",
1852+
Params: fftypes.FFIParams{
1853+
{
1854+
Name: "value",
1855+
Schema: fftypes.JSONAnyPtr(`{"type": "integer"}`),
1856+
},
1857+
},
1858+
},
1859+
},
1860+
Topic: "test-topic",
1861+
},
1862+
}
1863+
1864+
mbi.On("NormalizeContractLocation", context.Background(), blockchain.NormalizeListener, sub.Location).Return(sub.Location, nil)
1865+
mbi.On("GenerateEventSignature", context.Background(), mock.Anything).Return("changed", nil)
1866+
mbi.On("GenerateEventSignatureWithLocation", context.Background(), mock.Anything, mock.Anything).Return("0x123:changed", nil)
1867+
mdi.On("GetContractListeners", context.Background(), "ns1", mock.Anything).Return(nil, nil, nil)
1868+
mdi.On("InsertContractListener", context.Background(), &sub.ContractListener).Return(nil)
1869+
mbi.On("AddContractListener", context.Background(), &sub.ContractListener, "").Return(nil)
1870+
mdi.On("UpdateContractListener", context.Background(), "ns1", mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
16911871

16921872
_, err := cm.AddContractListener(context.Background(), sub)
16931873
assert.EqualError(t, err, "pop")
@@ -1732,6 +1912,7 @@ func TestAddContractAPIListener(t *testing.T) {
17321912
mdi.On("InsertContractListener", context.Background(), mock.MatchedBy(func(l *core.ContractListener) bool {
17331913
return *l.Filters[0].Interface.ID == *interfaceID && l.Filters[0].Event.Name == "changed" && l.Topic == "test-topic"
17341914
})).Return(nil)
1915+
mdi.On("UpdateContractListener", context.Background(), "ns1", mock.Anything, mock.Anything).Return(nil)
17351916

17361917
_, err := cm.AddContractAPIListener(context.Background(), "simple", "changed", listener)
17371918
assert.NoError(t, err)

0 commit comments

Comments
 (0)