diff --git a/common/closer.go b/common/closer.go
new file mode 100644
index 00000000..ea70a136
--- /dev/null
+++ b/common/closer.go
@@ -0,0 +1,246 @@
+package common
+
+import (
+	"github.com/hashicorp/go-multierror"
+	"github.com/pkg/errors"
+	"github.com/sasha-s/go-deadlock"
+	"golang.org/x/net/context"
+	"io"
+	"sync"
+	"sync/atomic"
+)
+
+var (
+	ErrAlreadyClosed = errors.New("already closed")
+)
+
+type Closer interface {
+	io.Closer
+	// OnClose Deprecate
+	// Deprecated
+	OnClose() <-chan struct{} // TODO: Move to AsyncCloser
+	AddCloseHook(name string) (<-chan struct{}, func(err error))
+	AddChildCloser(Closer) Closer
+	Err() error
+}
+
+// AsyncCloser is a Closer that can be closed asynchronously.
+// Deprecated
+type AsyncCloser interface {
+	Closer
+	// FinishClosing
+	// Deprecated
+	FinishClosing(error)
+}
+
+type hook struct {
+	name string
+	ch   chan error
+}
+
+type closer struct {
+	io.Closer
+	//ctx        context.Context
+	doneCh     <-chan struct{}
+	cancelFunc context.CancelFunc
+	closeFunc  func() error
+	closeOnce  sync.Once
+
+	parent Closer
+
+	err      atomic.Value
+	mu       deadlock.Mutex
+	children map[Closer]struct{}
+	hooks    map[*hook]struct{}
+}
+
+func newCloser(ctx context.Context) *closer {
+	ctx, cancel := context.WithCancel(ctx)
+	return &closer{
+		doneCh:     ctx.Done(),
+		cancelFunc: cancel,
+	}
+}
+
+func NewDefaultCloser() Closer {
+	return newCloser(context.Background())
+}
+
+func NewCloser(doneCh <-chan struct{}, cancelFunc context.CancelFunc, closeFunc func() error) Closer {
+	return &closer{
+		doneCh:     doneCh,
+		cancelFunc: cancelFunc,
+		closeFunc:  closeFunc,
+	}
+}
+
+func NewCloserWithFunc(f func() error) Closer {
+	c := newCloser(context.Background())
+	c.closeFunc = f
+	return c
+}
+
+func NewAsyncCloser(ctx context.Context) AsyncCloser {
+	return &asyncCloser{
+		closer: newCloser(ctx),
+		errCh:  make(chan error, 1),
+	}
+}
+
+func (c *closer) AddChildCloser(child Closer) Closer {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.children == nil {
+		c.children = make(map[Closer]struct{})
+	}
+	c.children[child] = struct{}{}
+	if childCloser, ok := child.(*closer); ok {
+		childCloser.parent = c
+	}
+	return child
+}
+
+func (c *closer) removeChild(child Closer) bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.children == nil {
+		return false
+	}
+	_, ok := c.children[child]
+	if ok {
+		delete(c.children, child)
+	}
+	return ok
+}
+
+func (c *closer) close() error {
+	if err := c.Err(); err != nil {
+		return err
+	}
+
+	if c.closeFunc != nil {
+		var err error
+		c.closeOnce.Do(func() {
+			err = c.closeFunc()
+		})
+		if err != nil {
+			return err
+		}
+	}
+
+	c.cancelFunc()
+
+	c.mu.Lock()
+	var err error
+	for h := range c.hooks {
+		e := <-h.ch
+		if e != nil {
+			err = multierror.Append(err, errors.Wrap(e, "failed to close hook: "+h.name))
+		} else {
+			delete(c.hooks, h)
+		}
+	}
+
+	// Copy the children to close, to avoid locking while closing them.
+	childrenToClose := make([]Closer, 0, len(c.children))
+	for child := range c.children {
+		childrenToClose = append(childrenToClose, child)
+	}
+	c.mu.Unlock()
+
+	// Close each child without holding the c.mu lock.
+	for _, child := range childrenToClose {
+		e := child.Close()
+		if e != nil {
+			err = multierror.Append(err, e)
+		}
+	}
+
+	// Now remove the child, reacquiring the lock to do so safely.
+	c.mu.Lock()
+	for _, child := range childrenToClose {
+		delete(c.children, child)
+	}
+	c.mu.Unlock()
+	return err
+}
+
+func (c *closer) Close() error {
+	if err := c.Err(); err != nil {
+		return err
+	}
+
+	err := c.close()
+
+	if err != nil {
+		c.err.Store(err)
+	} else {
+		c.err.Store(ErrAlreadyClosed)
+	}
+
+	if p, ok := c.parent.(*closer); ok {
+		p.removeChild(c)
+	}
+
+	return err
+}
+
+func (c *closer) OnClose() <-chan struct{} {
+	return c.doneCh
+}
+
+func (c *closer) AddCloseHook(name string) (<-chan struct{}, func(err error)) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	h := &hook{
+		name: name,
+		ch:   make(chan error),
+	}
+	if c.hooks == nil {
+		c.hooks = make(map[*hook]struct{})
+	}
+	c.hooks[h] = struct{}{}
+	return c.doneCh, func(err error) {
+		h.ch <- err
+		close(h.ch)
+	}
+}
+
+func (c *closer) Err() error {
+	e := c.err.Load()
+	if e == nil {
+		return nil
+	}
+	if !errors.Is(e.(error), ErrAlreadyClosed) {
+		return errors.Wrap(e.(error), "closer has been closed but failed")
+	}
+	return e.(error)
+}
+
+type asyncCloser struct {
+	*closer
+	errCh chan error
+}
+
+func (c *asyncCloser) Close() error {
+	err := c.closer.close()
+	if errors.Is(err, ErrAlreadyClosed) {
+		return ErrAlreadyClosed
+	}
+
+	asyncErr := <-c.errCh
+	if err != nil || asyncErr != nil {
+		err = multierror.Append(err, asyncErr)
+		c.err.Store(err)
+		return err
+	}
+	c.err.Store(ErrAlreadyClosed)
+	return nil
+}
+
+func (c *asyncCloser) FinishClosing(err error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.errCh <- err
+	close(c.errCh)
+}
diff --git a/common/closer_test.go b/common/closer_test.go
new file mode 100644
index 00000000..7feb34c4
--- /dev/null
+++ b/common/closer_test.go
@@ -0,0 +1,132 @@
+package common
+
+import (
+	"errors"
+	"github.com/stretchr/testify/assert"
+	"golang.org/x/net/context"
+	"log/slog"
+	"testing"
+)
+
+func TestCloser(t *testing.T) {
+	tests := []struct {
+		name      string
+		setup     func() (Closer, error)
+		checkFunc func(error, error) bool
+	}{
+		{
+			name: "Basic closer",
+			setup: func() (Closer, error) {
+				return NewDefaultCloser(), nil
+			},
+		},
+		{
+			name: "Async closer without error",
+			setup: func() (Closer, error) {
+				ac := NewAsyncCloser(context.Background())
+				go func() {
+					<-ac.OnClose()
+					ac.FinishClosing(nil)
+				}()
+				return ac, nil
+			},
+		},
+		{
+			name: "Async closer with error",
+			setup: func() (Closer, error) {
+				expectedErr := errors.New("expected error")
+				ac := NewAsyncCloser(context.Background())
+				go func() {
+					<-ac.OnClose()
+					ac.FinishClosing(expectedErr)
+				}()
+				return ac, expectedErr
+			},
+		},
+		{
+			name: "Multiple Closers",
+			setup: func() (Closer, error) {
+				c := NewDefaultCloser()
+				_ = c.AddChildCloser(NewDefaultCloser())
+				child2 := c.AddChildCloser(NewAsyncCloser(context.Background())).(AsyncCloser)
+				go func() {
+					<-child2.OnClose()
+					child2.FinishClosing(nil)
+				}()
+				return c, nil
+			},
+		},
+		{
+			name: "Multiple Closers with error",
+			setup: func() (Closer, error) {
+				c := NewDefaultCloser()
+				_ = c.AddChildCloser(NewDefaultCloser())
+				child2 := c.AddChildCloser(NewAsyncCloser(context.Background())).(AsyncCloser)
+				err := errors.New("expected error")
+				go func() {
+					<-child2.OnClose()
+					child2.FinishClosing(err)
+				}()
+				return c, err
+			},
+		},
+		{
+			name: "Multiple Closers with parent closer errors",
+			setup: func() (Closer, error) {
+				c := NewDefaultCloser()
+				child1 := c.AddChildCloser(NewDefaultCloser())
+				child2 := c.AddChildCloser(NewDefaultCloser())
+				err := errors.New("expected error")
+				c1, f1 := child1.AddCloseHook("c1")
+				go func() {
+					<-c1
+					f1(err)
+				}()
+				c2, f2 := child2.AddCloseHook("c2")
+				go func() {
+					<-c2
+					f2(nil)
+				}()
+				return c, err
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			closer, expectedErr := tc.setup()
+			assert.Nil(t, closer.Err())
+			err := closer.Close()
+			slog.Info("Closed", slog.Any("err", err))
+			assert.ErrorIs(t, err, expectedErr)
+		})
+	}
+}
+
+func TestCloserCloseMultipleTimes(t *testing.T) {
+	closer := NewDefaultCloser()
+	assert.Nil(t, closer.Close())
+	assert.Equal(t, ErrAlreadyClosed, closer.Close())
+	assert.Equal(t, ErrAlreadyClosed, closer.Err())
+}
+
+func TestCloserFunc(t *testing.T) {
+	c := NewCloserWithFunc(func() error {
+		return nil
+	})
+
+	assert.Nil(t, c.Err())
+	assert.Nil(t, c.Close())
+	assert.ErrorIs(t, c.Close(), ErrAlreadyClosed)
+}
+
+func TestCloserFuncError(t *testing.T) {
+	expectedErr := errors.New("expected error")
+	c := NewCloserWithFunc(func() error {
+		return expectedErr
+	})
+
+	assert.Nil(t, c.Err())
+	assert.Equal(t, expectedErr, c.Close())
+	assert.ErrorIs(t, c.Close(), expectedErr)
+}
diff --git a/common/lifecycle/lifecycle.go b/common/lifecycle/lifecycle.go
new file mode 100644
index 00000000..97bf8814
--- /dev/null
+++ b/common/lifecycle/lifecycle.go
@@ -0,0 +1,68 @@
+package lifecycle
+
+import (
+	"github.com/functionstream/function-stream/common"
+	"golang.org/x/net/context"
+)
+
+type Lifecycle struct {
+	closer    common.Closer
+	parent    *Lifecycle
+	ctx       context.Context
+	closeFunc func() error
+}
+
+type LifecycleOption func(*Lifecycle)
+
+func NewLifecycle(opts ...LifecycleOption) *Lifecycle {
+	l := &Lifecycle{}
+	for _, opt := range opts {
+		opt(l)
+	}
+	if l.closer == nil {
+		l.closer = common.NewCloser(l.ctx, l.closeFunc)
+	}
+	if l.parent != nil {
+		l.parent.closer.AddChildCloser(l.closer)
+	}
+	return l
+}
+
+func WithParent(parent *Lifecycle) LifecycleOption {
+	return func(l *Lifecycle) {
+		l.parent = parent
+		l.ctx = parent.ctx
+	}
+}
+
+func WithCloseFunc(f func() error) LifecycleOption {
+	return func(l *Lifecycle) {
+		l.closeFunc = f
+	}
+}
+
+func WithContext(ctx context.Context) LifecycleOption {
+	return func(l *Lifecycle) {
+		l.ctx = ctx
+	}
+}
+
+func (l *Lifecycle) GetLifecycle() *Lifecycle {
+	return l
+}
+
+func (l *Lifecycle) Context() context.Context {
+	return l.ctx
+}
+
+func (l *Lifecycle) AddCloseHook(name string) (<-chan struct{}, func(err error)) {
+	return l.closer.AddCloseHook(name)
+}
+
+func (l *Lifecycle) CheckState() error {
+	return l.closer.Err()
+}
+
+func (l *Lifecycle) Close() error {
+	return l.closer.Close()
+}
diff --git a/common/lifecycle/lifecycle_test.go b/common/lifecycle/lifecycle_test.go
new file mode 100644
index 00000000..19917571
--- /dev/null
+++ b/common/lifecycle/lifecycle_test.go
@@ -0,0 +1,27 @@
+package lifecycle
+
+import (
+	"github.com/functionstream/function-stream/common"
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestLifecycle_AddCloseHook(t *testing.T) {
+	l := NewLifecycle()
+	c, f := l.AddCloseHook("test")
+	err := errors.New("expected error")
+	go func() {
+		<-c
+		f(err)
+	}()
+	assert.ErrorIs(t, l.Close(), err)
+	assert.ErrorIs(t, l.CheckState(), err)
+}
+
+func TestLifecycle_CheckState(t *testing.T) {
+	l := NewLifecycle()
+	assert.NoError(t, l.CheckState())
+	assert.NoError(t, l.Close())
+	assert.ErrorIs(t, l.CheckState(), common.ErrAlreadyClosed)
+}
diff --git a/fs/api/instance.go b/fs/api/instance.go
index 6ffa7920..92deb118 100644
--- a/fs/api/instance.go
+++ b/fs/api/instance.go
@@ -17,18 +17,21 @@
 package api
 
 import (
+	"github.com/functionstream/function-stream/common/lifecycle"
 	"github.com/functionstream/function-stream/common/model"
 	"github.com/functionstream/function-stream/fs/contube"
 	"golang.org/x/net/context"
+	"io"
 	"log/slog"
 )
 
 type FunctionInstance interface {
+	io.Closer
+	GetLifecycle() *lifecycle.Lifecycle
 	Context() context.Context
 	FunctionContext() FunctionContext
 	Definition() *model.Function
 	Index() int32
-	Stop()
 	Run(factory FunctionRuntimeFactory)
 	WaitForReady() <-chan error
 	Logger() *slog.Logger
diff --git a/fs/api/runtime.go b/fs/api/runtime.go
index 55df3e8e..7df9657d 100644
--- a/fs/api/runtime.go
+++ b/fs/api/runtime.go
@@ -17,13 +17,15 @@
 package api
 
 import (
+	"github.com/functionstream/function-stream/common/lifecycle"
 	"github.com/functionstream/function-stream/fs/contube"
 )
 
 type FunctionRuntime interface {
+	GetLifecycle() *lifecycle.Lifecycle
 	WaitForReady() <-chan error
 	Call(e contube.Record) (contube.Record, error)
-	Stop()
+	Close() error
 }
 
 type FunctionRuntimeFactory interface {
diff --git a/fs/instance_impl.go b/fs/instance_impl.go
index 85767a51..0ad40e6c 100644
--- a/fs/instance_impl.go
+++ b/fs/instance_impl.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/functionstream/function-stream/common"
+	"github.com/functionstream/function-stream/common/lifecycle"
 	"github.com/functionstream/function-stream/common/model"
 	"github.com/functionstream/function-stream/fs/api"
 	"github.com/functionstream/function-stream/fs/contube"
@@ -28,6 +29,7 @@ import (
 )
 
 type FunctionInstanceImpl struct {
+	*lifecycle.Lifecycle
 	ctx           context.Context
 	funcCtx       api.FunctionContext
 	cancelFunc    context.CancelFunc
@@ -60,6 +62,7 @@ func (f *DefaultInstanceFactory) NewFunctionInstance(definition *model.Function,
 	ctx = context.WithValue(ctx, CtxKeyFunctionName, definition.Name)
 	ctx = context.WithValue(ctx, CtxKeyInstanceIndex, index)
 	return &FunctionInstanceImpl{
+		Lifecycle:     lifecycle.NewLifecycle(lifecycle.WithContext(ctx)),
 		ctx:           ctx,
 		funcCtx:       funcCtx,
 		cancelFunc:    cancelFunc,
@@ -75,10 +78,15 @@ func (f *DefaultInstanceFactory) NewFunctionInstance(definition *model.Function,
 
 func (instance *FunctionInstanceImpl) Run(runtimeFactory api.FunctionRuntimeFactory) {
 	runtime, err := runtimeFactory.NewFunctionRuntime(instance)
+	closeCh, closeF := instance.AddCloseHook("runtime")
 	if err != nil {
 		instance.readyCh <- errors.Wrap(err, "Error creating runtime")
 		return
 	}
+	defer func() {
+		e := runtime.Close()
+		closeF(e)
+	}()
 	getTubeConfig := func(config contube.ConfigMap, tubeConfig *model.TubeConfig) contube.ConfigMap {
 		if tubeConfig != nil && tubeConfig.Config != nil {
 			return contube.MergeConfig(config, tubeConfig.Config)
@@ -123,7 +131,7 @@ func (instance *FunctionInstanceImpl) Run(runtimeFactory api.FunctionRuntimeFact
 		}
 		select {
 		case sinkChan <- output:
-		case <-instance.ctx.Done():
+		case <-closeCh:
 			return
 		}
 
@@ -134,13 +142,14 @@ func (instance *FunctionInstanceImpl) WaitForReady() <-chan error {
 	return instance.readyCh
 }
 
-func (instance *FunctionInstanceImpl) Stop() {
+func (instance *FunctionInstanceImpl) Close() error {
 	instance.log.InfoContext(instance.ctx, "stopping function instance")
-	instance.cancelFunc()
-}
-
-func (instance *FunctionInstanceImpl) Context() context.Context {
-	return instance.ctx
+	err := instance.Lifecycle.Close()
+	if err != nil {
+		instance.log.Error("Error closing function instance", slog.Any("error", err))
+		return err
+	}
+	return nil
 }
 
 func (instance *FunctionInstanceImpl) FunctionContext() api.FunctionContext {
diff --git a/fs/manager.go b/fs/manager.go
index 2d659127..472f199b 100644
--- a/fs/manager.go
+++ b/fs/manager.go
@@ -24,6 +24,7 @@ import (
 	"github.com/functionstream/function-stream/fs/contube"
 	"github.com/functionstream/function-stream/fs/runtime/wazero"
 	"github.com/functionstream/function-stream/fs/statestore"
+	"github.com/hashicorp/go-multierror"
 	"github.com/pkg/errors"
 	"log/slog"
 	"math/rand"
@@ -207,7 +208,7 @@ func (fm *FunctionManager) StartFunction(f *model.Function) error {
 		case err := <-instance.WaitForReady():
 			if err != nil {
 				fm.log.ErrorContext(instance.Context(), "Error starting function instance", slog.Any("error", err.Error()))
-				instance.Stop()
+				_ = instance.Close()
 				return err
 			}
 		case <-instance.Context().Done():
@@ -226,10 +227,14 @@ func (fm *FunctionManager) DeleteFunction(name string) error {
 		return common.ErrorFunctionNotFound
 	}
 	delete(fm.functions, name)
+	var err error
 	for _, instance := range instances {
-		instance.Stop()
+		e := instance.Close()
+		if e != nil {
+			err = multierror.Append(err, e)
+		}
 	}
-	return nil
+	return err
 }
 
 func (fm *FunctionManager) ListFunctions() (result []string) {
@@ -274,14 +279,18 @@ func (fm *FunctionManager) GetStateStore() api.StateStore {
 func (fm *FunctionManager) Close() error {
 	fm.functionsLock.Lock()
 	defer fm.functionsLock.Unlock()
+	var err error
 	for _, instances := range fm.functions {
 		for _, instance := range instances {
-			instance.Stop()
+			e := instance.Close()
+			if e != nil {
+				err = multierror.Append(err, e)
+			}
 		}
 	}
-	err := fm.options.stateStore.Close()
-	if err != nil {
-		return err
+	e := fm.options.stateStore.Close()
+	if e != nil {
+		err = multierror.Append(err, e)
 	}
-	return nil
+	return err
 }
diff --git a/fs/runtime/grpc/grpc_func.go b/fs/runtime/grpc/grpc_func.go
index 25f81516..b6a36523 100644
--- a/fs/runtime/grpc/grpc_func.go
+++ b/fs/runtime/grpc/grpc_func.go
@@ -19,6 +19,7 @@ package grpc
 import (
 	"fmt"
 	"github.com/functionstream/function-stream/common"
+	"github.com/functionstream/function-stream/common/lifecycle"
 	"github.com/functionstream/function-stream/fs/api"
 	"github.com/functionstream/function-stream/fs/contube"
 	"github.com/functionstream/function-stream/fs/runtime/grpc/proto"
@@ -32,7 +33,7 @@ import (
 )
 
 type GRPCFuncRuntime struct {
-	api.FunctionRuntime
+	*lifecycle.Lifecycle
 	Name       string
 	instance   api.FunctionInstance
 	ctx        context.Context
@@ -148,11 +149,12 @@ func (s *FSSReconcileServer) NewFunctionRuntime(instance api.FunctionInstance) (
 	log := instance.Logger().With(
 		slog.String("component", "grpc-runtime"),
 	)
-	go func() {
-		<-instance.Context().Done()
-		s.removeFunction(name)
-	}()
 	runtime := &GRPCFuncRuntime{
+		Lifecycle: lifecycle.NewLifecycle(lifecycle.WithParent(instance.GetLifecycle()),
+			lifecycle.WithCloseFunc(func() error {
+				s.removeFunction(name)
+				return nil
+			})),
 		Name:     name,
 		instance: instance,
 		readyCh:  make(chan error),
@@ -163,9 +165,6 @@ func (s *FSSReconcileServer) NewFunctionRuntime(instance api.FunctionInstance) (
 			Status: proto.FunctionStatus_CREATING,
 		},
 		ctx: instance.Context(),
-		stopFunc: func() { // TODO: remove it, we should use instance.ctx to control the lifecycle
-			s.removeFunction(name)
-		},
 		log: log,
 	}
 	{
diff --git a/fs/runtime/wazero/wazero_runtime.go b/fs/runtime/wazero/wazero_runtime.go
index 8661b6c8..65b1d389 100644
--- a/fs/runtime/wazero/wazero_runtime.go
+++ b/fs/runtime/wazero/wazero_runtime.go
@@ -18,6 +18,7 @@ package wazero
 
 import (
 	"github.com/functionstream/function-stream/common"
+	"github.com/functionstream/function-stream/common/lifecycle"
 	"github.com/functionstream/function-stream/fs/api"
 	"github.com/functionstream/function-stream/fs/contube"
 	"github.com/pkg/errors"
@@ -84,6 +85,10 @@ func (f *WazeroFunctionRuntimeFactory) NewFunctionRuntime(instance api.FunctionI
 		return nil, errors.New("No process function found")
 	}
 	return &WazeroFunctionRuntime{
+		Lifecycle: lifecycle.NewLifecycle(lifecycle.WithParent(instance.GetLifecycle()),
+			lifecycle.WithCloseFunc(func() error {
+				return r.Close(instance.Context())
+			})),
 		callFunc: func(e contube.Record) (contube.Record, error) {
 			stdin.ResetBuffer(e.GetPayload())
 			_, err := process.Call(instance.Context())
@@ -93,20 +98,13 @@ func (f *WazeroFunctionRuntimeFactory) NewFunctionRuntime(instance api.FunctionI
 			output := stdout.GetAndReset()
 			return contube.NewRecordImpl(output, e.Commit), nil
 		},
-		stopFunc: func() {
-			err := r.Close(instance.Context())
-			if err != nil {
-				slog.ErrorContext(instance.Context(), "Error closing r", err)
-			}
-		},
 		log: log,
 	}, nil
 }
 
 type WazeroFunctionRuntime struct {
-	api.FunctionRuntime
+	*lifecycle.Lifecycle
 	callFunc func(e contube.Record) (contube.Record, error)
-	stopFunc func()
 	log      *slog.Logger
 }
 
@@ -119,7 +117,3 @@ func (r *WazeroFunctionRuntime) WaitForReady() <-chan error {
 func (r *WazeroFunctionRuntime) Call(e contube.Record) (contube.Record, error) {
 	return r.callFunc(e)
 }
-
-func (r *WazeroFunctionRuntime) Stop() {
-	r.stopFunc()
-}
diff --git a/go.mod b/go.mod
index 7bc614c1..3f5c8b8e 100644
--- a/go.mod
+++ b/go.mod
@@ -45,7 +45,7 @@ require (
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 // indirect
 	github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c // indirect
-	github.com/hashicorp/errwrap v1.0.0 // indirect
+	github.com/hashicorp/errwrap v1.1.0 // indirect
 	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
@@ -56,6 +56,7 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/mtibben/percent v0.2.1 // indirect
+	github.com/petermattis/goid v0.0.0-20180202154549-b0b1615b78e5 // indirect
 	github.com/pierrec/lz4 v2.6.1+incompatible // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_golang v1.19.0 // indirect
@@ -63,6 +64,7 @@ require (
 	github.com/prometheus/common v0.48.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/rogpeppe/go-internal v1.12.0 // indirect
+	github.com/sasha-s/go-deadlock v0.3.1 // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
diff --git a/go.sum b/go.sum
index 388522f4..988f7287 100644
--- a/go.sum
+++ b/go.sum
@@ -81,6 +81,8 @@ github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c h1:6rhixN/i8
 github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c/go.mod h1:NMPJylDgVpX0MLRlPy15sqSwOFv/U1GZ2m21JhFfek0=
 github.com/hashicorp/errwrap v1.0.0 h1:hLrqtEDnRye3+sgx6z4qVLNuviH3MR5aQ0ykNJa/UYA=
 github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
+github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I=
+github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
 github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo=
 github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
@@ -113,6 +115,8 @@ github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
 github.com/onsi/ginkgo v1.16.5/go.mod h1:+E8gABHa3K6zRBolWtd+ROzc/U5bkGt0FwiG042wbpU=
 github.com/onsi/gomega v1.29.0 h1:KIA/t2t5UBzoirT4H9tsML45GEbo3ouUnBHsCfD2tVg=
 github.com/onsi/gomega v1.29.0/go.mod h1:9sxs+SwGrKI0+PWe4Fxa9tFQQBG5xSsSbMXOI8PPpoQ=
+github.com/petermattis/goid v0.0.0-20180202154549-b0b1615b78e5 h1:q2e307iGHPdTGp0hoxKjt1H5pDo6utceo3dQVK3I5XQ=
+github.com/petermattis/goid v0.0.0-20180202154549-b0b1615b78e5/go.mod h1:jvVRKCrJTQWu0XVbaOlby/2lO20uSCHEMzzplHXte1o=
 github.com/pierrec/lz4 v2.6.1+incompatible h1:9UY3+iC23yxF0UfGaYrGplQ+79Rg+h/q9FV9ix19jjM=
 github.com/pierrec/lz4 v2.6.1+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
 github.com/pingcap/errors v0.11.4 h1:lFuQV/oaUMGcD2tqt+01ROSmJs75VG1ToEOkZIZ4nE4=
@@ -134,6 +138,8 @@ github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/f
 github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
 github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
+github.com/sasha-s/go-deadlock v0.3.1 h1:sqv7fDNShgjcaxkO0JNcOAlr8B9+cV5Ey/OB71efZx0=
+github.com/sasha-s/go-deadlock v0.3.1/go.mod h1:F73l+cr82YSh10GxyRI6qZiCgK64VaZjwesgfQ1/iLM=
 github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
 github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
 github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI=
diff --git a/server/server_test.go b/server/server_test.go
index 5732e4e2..20ed50c6 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -22,6 +22,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/functionstream/function-stream/common"
+	"github.com/functionstream/function-stream/common/lifecycle"
 	"github.com/functionstream/function-stream/common/model"
 	"github.com/functionstream/function-stream/fs"
 	"github.com/functionstream/function-stream/fs/api"
@@ -205,6 +206,7 @@ func (f *MockRuntimeFactory) NewFunctionRuntime(instance api.FunctionInstance) (
 }
 
 type MockRuntime struct {
+	*lifecycle.Lifecycle
 	funcCtx api.FunctionContext
 }