29667068588

Committed 19 Jul 2026 12:30AM UTC coverage: 82.155% (-0.005%) from 82.16%

Build # 29667068588

Build Type

push

github

Committed by

web-flow

Commit Message

Plugin fixes (#28)

* plugin fixes

* review, SERVER_DELETED before SERVER_POST_DELETE fix

* review, prev status fix

* review, limit async dispatches

* review, fix manager and plugin id

* review, busy plugin

* ci logging fixes

* plugin load fix

* fix status transitions

Coverage Stats

349 of 436 new or added lines in 24 files covered. (80.05%)

3 existing lines in 2 files now uncovered.

47365 of 57653 relevant lines covered (82.16%)

34725.02 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

64.97

/pkg/plugin/wrapper.go

package plugin

import (
        "context"
        "time"

        "github.com/gameap/gameap/pkg/plugin/proto"
        "github.com/pkg/errors"
        "github.com/tetratelabs/wazero/api"
)

// defaultCallTimeout caps guest calls whose caller did not set a deadline,
// so a runaway plugin cannot hold the per-plugin call gate forever.
const defaultCallTimeout = 30 * time.Second

// pluginServiceWrapper wraps WASM module calls to implement proto.PluginService.
type pluginServiceWrapper struct {
        // gate serializes guest calls; a channel instead of a mutex so queued
        // callers can abandon the wait when their context ends.
        gate                chan struct{}
        module              api.Module
        malloc              api.Function
        free                api.Function
        getinfo             api.Function
        initialize          api.Function
        shutdown            api.Function
        handleevent         api.Function
        getsubscribedevents api.Function
        gethttproutes       api.Function
        handlehttprequest   api.Function
        getfrontendbundle   api.Function
        getserverabilities  api.Function
}

func (p *pluginServiceWrapper) callFunction(
        ctx context.Context,
        fn api.Function,
        request vtMarshaler,
) ([]byte, error) {
        // The wait honors the caller's full context (deadline and cancellation):
        // the guest has not been invoked yet, so giving up here is always safe.
        select {
        case p.gate <- struct{}{}:
        case <-ctx.Done():
                return nil, errors.Wrapf(ErrPluginBusy, "%s", ctx.Err())
        }
        defer func() { <-p.gate }()

        // select picks randomly when both cases are ready.
        if ctx.Err() != nil {
                return nil, errors.Wrapf(ErrPluginBusy, "%s", ctx.Err())
        }

        // The runtime closes the module when the call context is done
        // (WithCloseOnContextDone), so caller cancellation (e.g. a client
        // dropping an HTTP request) must not reach the guest — only explicit
        // deadlines may interrupt it.
        var cancel context.CancelFunc
        if deadline, ok := ctx.Deadline(); ok {
                ctx, cancel = context.WithDeadline(context.WithoutCancel(ctx), deadline)
        } else {
                ctx, cancel = context.WithTimeout(context.WithoutCancel(ctx), defaultCallTimeout)
        }
        defer cancel()

        data, err := request.MarshalVT()
        if err != nil {
                return nil, err
        }

        dataSize := uint64(len(data))

        var dataPtr uint64
        if dataSize != 0 {
                results, callErr := p.malloc.Call(ctx, dataSize)
                if callErr != nil {
                        return nil, callErr
                }

                dataPtr = results[0]
                defer p.free.Call(ctx, dataPtr) //nolint:errcheck

                if !p.module.Memory().Write(uint32(dataPtr), data) { //nolint:gosec
                        return nil, errors.Wrapf(ErrMemoryOutOfRange, "write(%d, %d), size=%d",
                                dataPtr, dataSize, p.module.Memory().Size())
                }
        }

        ptrSize, err := fn.Call(ctx, dataPtr, dataSize)
        if err != nil {
                return nil, err
        }

        resPtr := uint32(ptrSize[0] >> 32)
        resSize := uint32(ptrSize[0]) //nolint:gosec
        isErrResponse := (resSize & (1 << 31)) > 0

        if isErrResponse {
                resSize &^= (1 << 31)
        }

        if resPtr != 0 {
                defer p.free.Call(ctx, uint64(resPtr)) //nolint:errcheck
        }

        bytes, ok := p.module.Memory().Read(resPtr, resSize)
        if !ok {
                return nil, errors.Wrapf(ErrMemoryOutOfRange, "read(%d, %d), size=%d",
                        resPtr, resSize, p.module.Memory().Size())
        }

        if isErrResponse {
                return nil, errors.WithMessage(ErrPluginReturnedError, string(bytes))
        }

        return bytes, nil
}

type vtMarshaler interface {
        MarshalVT() ([]byte, error)
}

func (p *pluginServiceWrapper) GetInfo(
        ctx context.Context,
        request *proto.GetInfoRequest,
) (*proto.PluginInfo, error) {
        bytes, err := p.callFunction(ctx, p.getinfo, request)
        if err != nil {
                return nil, err
        }

        response := new(proto.PluginInfo)
        if err = response.UnmarshalVT(bytes); err != nil {
                return nil, err
        }

        return response, nil
}

func (p *pluginServiceWrapper) Initialize(
        ctx context.Context,
        request *proto.InitializeRequest,
) (*proto.InitializeResponse, error) {
        bytes, err := p.callFunction(ctx, p.initialize, request)
        if err != nil {
                return nil, err
        }

        response := new(proto.InitializeResponse)
        if err = response.UnmarshalVT(bytes); err != nil {
                return nil, err
        }

        return response, nil
}

func (p *pluginServiceWrapper) Shutdown(
        ctx context.Context,
        request *proto.ShutdownRequest,
) (*proto.ShutdownResponse, error) {
        bytes, err := p.callFunction(ctx, p.shutdown, request)
        if err != nil {
                return nil, err
        }

        response := new(proto.ShutdownResponse)
        if err = response.UnmarshalVT(bytes); err != nil {
                return nil, err
        }

        return response, nil
}

func (p *pluginServiceWrapper) HandleEvent(
        ctx context.Context,
        request *proto.Event,
) (*proto.EventResult, error) {
        bytes, err := p.callFunction(ctx, p.handleevent, request)
        if err != nil {
                return nil, err
        }

        response := new(proto.EventResult)
        if err = response.UnmarshalVT(bytes); err != nil {
                return nil, err
        }

        return response, nil
}

func (p *pluginServiceWrapper) GetSubscribedEvents(
        ctx context.Context,
        request *proto.GetSubscribedEventsRequest,
) (*proto.GetSubscribedEventsResponse, error) {
        bytes, err := p.callFunction(ctx, p.getsubscribedevents, request)
        if err != nil {
                return nil, err
        }

        response := new(proto.GetSubscribedEventsResponse)
        if err = response.UnmarshalVT(bytes); err != nil {
                return nil, err
        }

        return response, nil
}

func (p *pluginServiceWrapper) GetHTTPRoutes(
        ctx context.Context,
        request *proto.GetHTTPRoutesRequest,
) (*proto.GetHTTPRoutesResponse, error) {
        bytes, err := p.callFunction(ctx, p.gethttproutes, request)
        if err != nil {
                return nil, err
        }

        response := new(proto.GetHTTPRoutesResponse)
        if err = response.UnmarshalVT(bytes); err != nil {
                return nil, err
        }

        return response, nil
}

func (p *pluginServiceWrapper) HandleHTTPRequest(
        ctx context.Context,
        request *proto.HTTPRequest,
) (*proto.HTTPResponse, error) {
        bytes, err := p.callFunction(ctx, p.handlehttprequest, request)
        if err != nil {
                return nil, err
        }

        response := new(proto.HTTPResponse)
        if err = response.UnmarshalVT(bytes); err != nil {
                return nil, err
        }

        return response, nil
}

func (p *pluginServiceWrapper) GetFrontendBundle(
        ctx context.Context,
        request *proto.GetFrontendBundleRequest,
) (*proto.GetFrontendBundleResponse, error) {
        if p.getfrontendbundle == nil {
                return &proto.GetFrontendBundleResponse{HasBundle: false}, nil
        }

        bytes, err := p.callFunction(ctx, p.getfrontendbundle, request)
        if err != nil {
                return nil, err
        }

        response := new(proto.GetFrontendBundleResponse)
        if err = response.UnmarshalVT(bytes); err != nil {
                return nil, err
        }

        return response, nil
}

func (p *pluginServiceWrapper) GetServerAbilities(
        ctx context.Context,
        request *proto.GetServerAbilitiesRequest,
) (*proto.GetServerAbilitiesResponse, error) {
        if p.getserverabilities == nil {
                return &proto.GetServerAbilitiesResponse{Abilities: nil}, nil
        }

        bytes, err := p.callFunction(ctx, p.getserverabilities, request)
        if err != nil {
                return nil, err
        }

        response := new(proto.GetServerAbilitiesResponse)
        if err = response.UnmarshalVT(bytes); err != nil {
                return nil, err
        }

        return response, nil
}

1	package plugin
2
3	import (
4	"context"
5	"time"
6
7	"github.com/gameap/gameap/pkg/plugin/proto"
8	"github.com/pkg/errors"
9	"github.com/tetratelabs/wazero/api"
10	)
11
12	// defaultCallTimeout caps guest calls whose caller did not set a deadline,
13	// so a runaway plugin cannot hold the per-plugin call gate forever.
14	const defaultCallTimeout = 30 * time.Second
15
16	// pluginServiceWrapper wraps WASM module calls to implement proto.PluginService.
17	type pluginServiceWrapper struct {
18	// gate serializes guest calls; a channel instead of a mutex so queued
19	// callers can abandon the wait when their context ends.
20	gate chan struct{}
21	module api.Module
22	malloc api.Function
23	free api.Function
24	getinfo api.Function
25	initialize api.Function
26	shutdown api.Function
27	handleevent api.Function
28	getsubscribedevents api.Function
29	gethttproutes api.Function
30	handlehttprequest api.Function
31	getfrontendbundle api.Function
32	getserverabilities api.Function
33	}
34
35	func (p *pluginServiceWrapper) callFunction(
36	ctx context.Context,
37	fn api.Function,
38	request vtMarshaler,
39	) ([]byte, error) {	29✔
40	// The wait honors the caller's full context (deadline and cancellation):	29✔
41	// the guest has not been invoked yet, so giving up here is always safe.	29✔
42	select {	29✔
43	case p.gate <- struct{}{}:	27✔
44	case <-ctx.Done():	2✔
45	return nil, errors.Wrapf(ErrPluginBusy, "%s", ctx.Err())	2✔
46	}
47	defer func() { <-p.gate }()	54✔
48
49	// select picks randomly when both cases are ready.
50	if ctx.Err() != nil {	27✔
NEW 51	return nil, errors.Wrapf(ErrPluginBusy, "%s", ctx.Err())	×
NEW 52	}	×
53
54	// The runtime closes the module when the call context is done
55	// (WithCloseOnContextDone), so caller cancellation (e.g. a client
56	// dropping an HTTP request) must not reach the guest — only explicit
57	// deadlines may interrupt it.
58	var cancel context.CancelFunc	27✔
59	if deadline, ok := ctx.Deadline(); ok {	27✔
NEW 60	ctx, cancel = context.WithDeadline(context.WithoutCancel(ctx), deadline)	×
61	} else {	27✔
62	ctx, cancel = context.WithTimeout(context.WithoutCancel(ctx), defaultCallTimeout)	27✔
63	}	27✔
64	defer cancel()	27✔
65		27✔
66	data, err := request.MarshalVT()	27✔
67	if err != nil {	27✔
68	return nil, err	×
69	}	×
70
71	dataSize := uint64(len(data))	27✔
72		27✔
73	var dataPtr uint64	27✔
74	if dataSize != 0 {	36✔
75	results, callErr := p.malloc.Call(ctx, dataSize)	9✔
76	if callErr != nil {	9✔
77	return nil, callErr	×
78	}	×
79
80	dataPtr = results[0]	9✔
81	defer p.free.Call(ctx, dataPtr) //nolint:errcheck	9✔
82		9✔
83	if !p.module.Memory().Write(uint32(dataPtr), data) { //nolint:gosec	9✔
84	return nil, errors.Wrapf(ErrMemoryOutOfRange, "write(%d, %d), size=%d",	×
85	dataPtr, dataSize, p.module.Memory().Size())	×
86	}	×
87	}
88
89	ptrSize, err := fn.Call(ctx, dataPtr, dataSize)	27✔
90	if err != nil {	27✔
91	return nil, err	×
92	}	×
93
94	resPtr := uint32(ptrSize[0] >> 32)	27✔
95	resSize := uint32(ptrSize[0]) //nolint:gosec	27✔
96	isErrResponse := (resSize & (1 << 31)) > 0	27✔
97		27✔
98	if isErrResponse {	27✔
99	resSize &^= (1 << 31)	×
100	}	×
101
102	if resPtr != 0 {	53✔
103	defer p.free.Call(ctx, uint64(resPtr)) //nolint:errcheck	26✔
104	}	26✔
105
106	bytes, ok := p.module.Memory().Read(resPtr, resSize)	27✔
107	if !ok {	27✔
108	return nil, errors.Wrapf(ErrMemoryOutOfRange, "read(%d, %d), size=%d",	×
109	resPtr, resSize, p.module.Memory().Size())	×
110	}	×
111
112	if isErrResponse {	27✔
113	return nil, errors.WithMessage(ErrPluginReturnedError, string(bytes))	×
114	}	×
115
116	return bytes, nil	27✔
117	}
118
119	type vtMarshaler interface {
120	MarshalVT() ([]byte, error)
121	}
122
123	func (p *pluginServiceWrapper) GetInfo(
124	ctx context.Context,
125	request *proto.GetInfoRequest,
126	) (*proto.PluginInfo, error) {	5✔
127	bytes, err := p.callFunction(ctx, p.getinfo, request)	5✔
128	if err != nil {	5✔
129	return nil, err	×
130	}	×
131
132	response := new(proto.PluginInfo)	5✔
133	if err = response.UnmarshalVT(bytes); err != nil {	5✔
134	return nil, err	×
135	}	×
136
137	return response, nil	5✔
138	}
139
140	func (p *pluginServiceWrapper) Initialize(
141	ctx context.Context,
142	request *proto.InitializeRequest,
143	) (*proto.InitializeResponse, error) {	4✔
144	bytes, err := p.callFunction(ctx, p.initialize, request)	4✔
145	if err != nil {	4✔
146	return nil, err	×
147	}	×
148
149	response := new(proto.InitializeResponse)	4✔
150	if err = response.UnmarshalVT(bytes); err != nil {	4✔
151	return nil, err	×
152	}	×
153
154	return response, nil	4✔
155	}
156
157	func (p *pluginServiceWrapper) Shutdown(
158	ctx context.Context,
159	request *proto.ShutdownRequest,
160	) (*proto.ShutdownResponse, error) {	1✔
161	bytes, err := p.callFunction(ctx, p.shutdown, request)	1✔
162	if err != nil {	1✔
163	return nil, err	×
164	}	×
165
166	response := new(proto.ShutdownResponse)	1✔
167	if err = response.UnmarshalVT(bytes); err != nil {	1✔
168	return nil, err	×
169	}	×
170
171	return response, nil	1✔
172	}
173
174	func (p *pluginServiceWrapper) HandleEvent(
175	ctx context.Context,
176	request *proto.Event,
177	) (*proto.EventResult, error) {	2✔
178	bytes, err := p.callFunction(ctx, p.handleevent, request)	2✔
179	if err != nil {	2✔
180	return nil, err	×
181	}	×
182
183	response := new(proto.EventResult)	2✔
184	if err = response.UnmarshalVT(bytes); err != nil {	2✔
185	return nil, err	×
186	}	×
187
188	return response, nil	2✔
189	}
190
191	func (p *pluginServiceWrapper) GetSubscribedEvents(
192	ctx context.Context,
193	request *proto.GetSubscribedEventsRequest,
194	) (*proto.GetSubscribedEventsResponse, error) {	1✔
195	bytes, err := p.callFunction(ctx, p.getsubscribedevents, request)	1✔
196	if err != nil {	1✔
197	return nil, err	×
198	}	×
199
200	response := new(proto.GetSubscribedEventsResponse)	1✔
201	if err = response.UnmarshalVT(bytes); err != nil {	1✔
202	return nil, err	×
203	}	×
204
205	return response, nil	1✔
206	}
207
208	func (p *pluginServiceWrapper) GetHTTPRoutes(
209	ctx context.Context,
210	request *proto.GetHTTPRoutesRequest,
211	) (*proto.GetHTTPRoutesResponse, error) {	4✔
212	bytes, err := p.callFunction(ctx, p.gethttproutes, request)	4✔
213	if err != nil {	4✔
214	return nil, err	×
215	}	×
216
217	response := new(proto.GetHTTPRoutesResponse)	4✔
218	if err = response.UnmarshalVT(bytes); err != nil {	4✔
219	return nil, err	×
220	}	×
221
222	return response, nil	4✔
223	}
224
225	func (p *pluginServiceWrapper) HandleHTTPRequest(
226	ctx context.Context,
227	request *proto.HTTPRequest,
228	) (*proto.HTTPResponse, error) {	2✔
229	bytes, err := p.callFunction(ctx, p.handlehttprequest, request)	2✔
230	if err != nil {	2✔
231	return nil, err	×
232	}	×
233
234	response := new(proto.HTTPResponse)	2✔
235	if err = response.UnmarshalVT(bytes); err != nil {	2✔
236	return nil, err	×
237	}	×
238
239	return response, nil	2✔
240	}
241
242	func (p *pluginServiceWrapper) GetFrontendBundle(
243	ctx context.Context,
244	request *proto.GetFrontendBundleRequest,
245	) (*proto.GetFrontendBundleResponse, error) {	5✔
246	if p.getfrontendbundle == nil {	6✔
247	return &proto.GetFrontendBundleResponse{HasBundle: false}, nil	1✔
248	}	1✔
249
250	bytes, err := p.callFunction(ctx, p.getfrontendbundle, request)	4✔
251	if err != nil {	4✔
252	return nil, err	×
253	}	×
254
255	response := new(proto.GetFrontendBundleResponse)	4✔
256	if err = response.UnmarshalVT(bytes); err != nil {	4✔
257	return nil, err	×
258	}	×
259
260	return response, nil	4✔
261	}
262
263	func (p *pluginServiceWrapper) GetServerAbilities(
264	ctx context.Context,
265	request *proto.GetServerAbilitiesRequest,
266	) (*proto.GetServerAbilitiesResponse, error) {	5✔
267	if p.getserverabilities == nil {	6✔
268	return &proto.GetServerAbilitiesResponse{Abilities: nil}, nil	1✔
269	}	1✔
270
271	bytes, err := p.callFunction(ctx, p.getserverabilities, request)	4✔
272	if err != nil {	4✔
273	return nil, err	×
274	}	×
275
276	response := new(proto.GetServerAbilitiesResponse)	4✔
277	if err = response.UnmarshalVT(bytes); err != nil {	4✔
278	return nil, err	×
279	}	×
280
281	return response, nil	4✔
282	}

gameap / gameap / 29667068588

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous