| // Copyright 2015 The Vanadium Authors. All rights reserved. |
| // Use of this source code is governed by a BSD-style |
| // license that can be found in the LICENSE file. |
| |
| package server |
| |
| import ( |
| "math/rand" |
| "path/filepath" |
| "sync" |
| "time" |
| |
| "v.io/v23/context" |
| "v.io/v23/glob" |
| "v.io/v23/query/engine" |
| ds "v.io/v23/query/engine/datasource" |
| "v.io/v23/query/syncql" |
| "v.io/v23/rpc" |
| "v.io/v23/security" |
| "v.io/v23/security/access" |
| wire "v.io/v23/services/syncbase" |
| pubutil "v.io/v23/syncbase/util" |
| "v.io/v23/verror" |
| "v.io/v23/vom" |
| "v.io/x/lib/vlog" |
| "v.io/x/ref/services/syncbase/common" |
| "v.io/x/ref/services/syncbase/server/interfaces" |
| "v.io/x/ref/services/syncbase/server/util" |
| "v.io/x/ref/services/syncbase/store" |
| storeutil "v.io/x/ref/services/syncbase/store/util" |
| "v.io/x/ref/services/syncbase/store/watchable" |
| "v.io/x/ref/services/syncbase/vsync" |
| sbwatchable "v.io/x/ref/services/syncbase/watchable" |
| ) |
| |
| // database is a per-database singleton (i.e. not per-request) that handles |
| // Database RPCs. |
| // Note: If a database does not exist at the time of a database RPC, the |
| // dispatcher creates a short-lived database object to service that particular |
| // request. |
| type database struct { |
| id wire.Id |
| s *service |
| // The fields below are initialized iff this database exists. |
| exists bool |
| // TODO(sadovsky): Make st point to a store.Store wrapper that handles paging, |
| // and do not actually open the store in NewDatabase. |
| st *watchable.Store // stores all data for a single database |
| |
| // Active snapshots and transactions corresponding to client batches. |
| // TODO(sadovsky): Add timeouts and GC. |
| mu sync.Mutex // protects the fields below |
| sns map[uint64]store.Snapshot |
| txs map[uint64]*transactionState |
| |
| // Active ConflictResolver connection from the app to this database. |
| // NOTE: For now, we assume there's only one open conflict resolution stream |
| // per database (typically, from the app that owns the database). |
| crStream wire.ConflictManagerStartConflictResolverServerCall |
| // Mutex lock to protect concurrent read/write of crStream pointer |
| crMu sync.Mutex |
| } |
| |
| var ( |
| _ wire.DatabaseServerMethods = (*database)(nil) |
| _ interfaces.Database = (*database)(nil) |
| ) |
| |
| // DatabaseOptions configures a database. |
| type DatabaseOptions struct { |
| // Database-level permissions. |
| Perms access.Permissions |
| // Root dir for data storage. This path is relative from the service's RootDir. |
| RootDir string |
| // Storage engine to use. |
| Engine string |
| } |
| |
| type permissionState struct { |
| dataChanged bool |
| permsChanged bool |
| initialPerms access.Permissions |
| finalPerms access.Permissions |
| } |
| |
| type transactionState struct { |
| tx *watchable.Transaction |
| permsChanges map[wire.Id]*permissionState |
| } |
| |
| // Keeps track that this collection had a mutation and the permissions at the time. If no |
| // permissions are yet known for the collection then remember the current permissions as the |
| // initial permissions of the collection. |
| // When the transaction is committed we will know to validate this collection's permissions. |
| func (ts *transactionState) MarkDataChanged(collectionId wire.Id, perms access.Permissions) { |
| state := ts.permsState(collectionId) |
| state.dataChanged = true |
| if state.initialPerms == nil { |
| state.initialPerms = perms |
| } |
| state.finalPerms = perms |
| } |
| |
| // Keeps track that the permissions were changed on this collection and the before and after |
| // permissions. If no permissions are yet known for the collection then remember the current |
| // permissions as the initial permissions of the collection. |
| func (ts *transactionState) MarkPermsChanged(collectionId wire.Id, permsBefore access.Permissions, permsAfter access.Permissions) { |
| state := ts.permsState(collectionId) |
| state.permsChanged = true |
| if state.initialPerms == nil { |
| state.initialPerms = permsBefore |
| } |
| state.finalPerms = permsAfter |
| } |
| |
| // Resets all tracked changes to the collection. Used on collection destroy. Since destroy |
| // cannot happen on a synced collection, the destroy and any updates before it will not be |
| // seen remotely, so validation must start from the implicit permissions in case the |
| // collection is created again. This also allows destroy to not require both write and |
| // admin permissions. |
| func (ts *transactionState) ResetCollectionChanges(collectionId wire.Id) { |
| delete(ts.permsChanges, collectionId) |
| } |
| |
| // validatePermissionChanges performs an auth check on each collection that has a data change or |
| // permission change and returns false if any of the auth checks fail. |
| // TODO(ivanpi): This check should be done against signing blessings at signing time, in |
| // both batch and non-batch cases. |
| // Note, service and database ACLs are not synced, so they don't need to be rechecked. |
| func (ts *transactionState) validatePermissionChanges(ctx *context.T, securityCall security.Call) bool { |
| for _, collectionState := range ts.permsChanges { |
| // This collection was modified, make sure that the write acl is either present at |
| // the end or that it had the write acl to begin with. This way we can be sure that |
| // a mutation didn't take place when it appeared that there was no write acl before |
| // and after the transaction. |
| if collectionState.dataChanged { |
| before := hasPermission(ctx, securityCall, collectionState.initialPerms, access.Write) |
| after := hasPermission(ctx, securityCall, collectionState.finalPerms, access.Write) |
| if !after && !before { |
| return false |
| } |
| } |
| |
| // The permissions were changed on the collection, make sure that the admin acl is |
| // present at the beginning. |
| if collectionState.permsChanged { |
| if !hasPermission(ctx, securityCall, collectionState.initialPerms, access.Admin) { |
| return false |
| } |
| } |
| } |
| return true |
| } |
| |
| func (ts *transactionState) permsState(collectionId wire.Id) *permissionState { |
| if ts.permsChanges == nil { |
| ts.permsChanges = make(map[wire.Id]*permissionState) |
| } |
| state, ok := ts.permsChanges[collectionId] |
| if !ok { |
| state = &permissionState{} |
| ts.permsChanges[collectionId] = state |
| } |
| return state |
| } |
| |
| // hasPermission returns true if the caller is authorized for the specific tag based on the |
| // passed in perms. |
| func hasPermission(ctx *context.T, securityCall security.Call, perms access.Permissions, tag access.Tag) bool { |
| // Authorize returns either an error or nil, so nil means the caller is authorized. |
| return common.TagAuthorizer(tag, perms).Authorize(ctx, securityCall) == nil |
| } |
| |
| // openDatabase opens a database and returns a *database for it. Designed for |
| // use from within newDatabase and newService. |
| func openDatabase(ctx *context.T, s *service, id wire.Id, opts DatabaseOptions, openOpts storeutil.OpenOptions) (*database, error) { |
| // DatabaseOption's RootDir is relative to the service's RootDir (but for backwards compatibility |
| // s.absRootDir will return any absolute paths as-is). |
| p := s.absRootDir(filepath.Join(opts.RootDir, opts.Engine)) |
| st, err := storeutil.OpenStore(opts.Engine, p, openOpts) |
| if err != nil { |
| return nil, err |
| } |
| wst, err := watchable.Wrap(st, s.vclock, &watchable.Options{ |
| // TODO(ivanpi): Since ManagedPrefixes control what gets synced, they |
| // should be moved to a more visible place (e.g. constants). Also consider |
| // decoupling managed and synced prefixes. |
| ManagedPrefixes: []string{common.CollectionPermsPrefix, common.RowPrefix}, |
| }) |
| if err != nil { |
| return nil, err |
| } |
| return &database{ |
| id: id, |
| s: s, |
| exists: true, |
| st: wst, |
| sns: make(map[uint64]store.Snapshot), |
| txs: make(map[uint64]*transactionState), |
| }, nil |
| } |
| |
| // newDatabase creates a new database instance and returns it. |
| // Designed for use from within service.createDatabase. |
| func newDatabase(ctx *context.T, s *service, id wire.Id, metadata *wire.SchemaMetadata, opts DatabaseOptions) (*database, error) { |
| if opts.Perms == nil { |
| return nil, verror.New(verror.ErrInternal, ctx, "perms must be specified") |
| } |
| d, err := openDatabase(ctx, s, id, opts, storeutil.OpenOptions{CreateIfMissing: true, ErrorIfExists: true}) |
| if err != nil { |
| return nil, err |
| } |
| data := &DatabaseData{ |
| Perms: opts.Perms, |
| SchemaMetadata: metadata, |
| } |
| if err := store.Put(ctx, d.st, d.stKey(), data); err != nil { |
| return nil, err |
| } |
| |
| // Start a Sync watcher on this newly created database store. |
| vsync.NewSyncDatabase(d).StartStoreWatcher(ctx) |
| |
| return d, nil |
| } |
| |
| //////////////////////////////////////// |
| // RPC methods |
| |
| func (d *database) Create(ctx *context.T, call rpc.ServerCall, metadata *wire.SchemaMetadata, perms access.Permissions) error { |
| // Permissions and existence checked in d.s.createDatabase. |
| return d.s.createDatabase(ctx, call, d.id, perms, metadata) |
| } |
| |
| func (d *database) Destroy(ctx *context.T, call rpc.ServerCall) error { |
| // Permissions and existence checked in d.s.destroyDatabase. |
| return d.s.destroyDatabase(ctx, call, d.id) |
| } |
| |
| func (d *database) Exists(ctx *context.T, call rpc.ServerCall) (bool, error) { |
| impl := func(sntx store.SnapshotOrTransaction) error { |
| _, _, err := d.GetDataWithExistAuth(ctx, call, sntx, &DatabaseData{}) |
| return err |
| } |
| return common.ErrorToExists(d.runWithNewSnapshot(ctx, call, impl)) |
| } |
| |
| var rng *rand.Rand = rand.New(rand.NewSource(time.Now().UTC().UnixNano())) |
| |
| func (d *database) BeginBatch(ctx *context.T, call rpc.ServerCall, opts wire.BatchOptions) (wire.BatchHandle, error) { |
| allowBeginBatch := wire.AllDatabaseTags |
| |
| if !d.exists { |
| return "", d.fuzzyNoExistError(ctx, call) |
| } |
| if _, err := common.GetPermsWithAuth(ctx, call, d, allowBeginBatch, d.st); err != nil { |
| return "", err |
| } |
| d.mu.Lock() |
| defer d.mu.Unlock() |
| var id uint64 |
| var batchType common.BatchType |
| for { |
| id = uint64(rng.Int63()) |
| if opts.ReadOnly { |
| if _, ok := d.sns[id]; !ok { |
| d.sns[id] = d.st.NewSnapshot() |
| batchType = common.BatchTypeSn |
| break |
| } |
| } else { |
| if _, ok := d.txs[id]; !ok { |
| d.txs[id] = &transactionState{tx: d.st.NewWatchableTransaction()} |
| batchType = common.BatchTypeTx |
| break |
| } |
| } |
| } |
| return common.JoinBatchHandle(batchType, id), nil |
| } |
| |
| func (d *database) Commit(ctx *context.T, call rpc.ServerCall, bh wire.BatchHandle) error { |
| allowCommit := wire.AllDatabaseTags |
| |
| if bh == "" { |
| return wire.NewErrNotBoundToBatch(ctx) |
| } |
| if !d.exists { |
| return d.fuzzyNoExistError(ctx, call) |
| } |
| _, ts, batchId, err := d.batchLookupInternal(ctx, call, bh) |
| if err != nil { |
| return err |
| } |
| if _, err := common.GetPermsWithAuth(ctx, call, d, allowCommit, d.st); err != nil { |
| return err |
| } |
| if ts == nil { |
| return wire.NewErrReadOnlyBatch(ctx) |
| } |
| if !ts.validatePermissionChanges(ctx, call.Security()) { |
| return wire.NewErrInvalidPermissionsChange(ctx) |
| } |
| if err = ts.tx.Commit(); err == nil { |
| d.mu.Lock() |
| delete(d.txs, batchId) |
| d.mu.Unlock() |
| } |
| // TODO(ivanpi): Best effort abort if commit fails? Watchable Commit can fail |
| // before the underlying snapshot is aborted. |
| if verror.ErrorID(err) == store.ErrConcurrentTransaction.ID { |
| return verror.New(wire.ErrConcurrentBatch, ctx, err) |
| } |
| return err |
| } |
| |
| func (d *database) Abort(ctx *context.T, call rpc.ServerCall, bh wire.BatchHandle) error { |
| allowAbort := wire.AllDatabaseTags |
| |
| if bh == "" { |
| return wire.NewErrNotBoundToBatch(ctx) |
| } |
| if !d.exists { |
| return d.fuzzyNoExistError(ctx, call) |
| } |
| sn, ts, batchId, err := d.batchLookupInternal(ctx, call, bh) |
| if err != nil { |
| return err |
| } |
| if _, err := common.GetPermsWithAuth(ctx, call, d, allowAbort, d.st); err != nil { |
| return err |
| } |
| if ts != nil { |
| d.mu.Lock() |
| delete(d.txs, batchId) |
| d.mu.Unlock() |
| // TODO(ivanpi): If tx.Abort fails, retry later? |
| return ts.tx.Abort() |
| } else { |
| d.mu.Lock() |
| delete(d.sns, batchId) |
| d.mu.Unlock() |
| // TODO(ivanpi): If sn.Abort fails, retry later? |
| return sn.Abort() |
| } |
| } |
| |
| func (d *database) Exec(ctx *context.T, call wire.DatabaseExecServerCall, bh wire.BatchHandle, q string, params []*vom.RawBytes) error { |
| // Permissions and existence checked in qdb.GetTable. |
| // RunInTransaction() cannot be used here because we may or may not be |
| // creating a transaction. qe.Exec must be called and the statement must be |
| // parsed before we know if a snapshot or a transaction should be created. To |
| // duplicate the semantics of RunInTransaction, if we are not inside a batch, |
| // we attempt the Exec up to 100 times and retry on ErrConcurrentTransaction. |
| // TODO(ivanpi): Refactor query parsing into a separate step, simplify request |
| // handling. Consider separate Query and Exec methods. |
| maxAttempts := 100 |
| attempt := 0 |
| for { |
| err := d.execInternal(ctx, call, bh, q, params) |
| if bh != "" || attempt >= maxAttempts || verror.ErrorID(err) != store.ErrConcurrentTransaction.ID { |
| return err |
| } |
| attempt++ |
| } |
| } |
| |
| func (d *database) execInternal(ctx *context.T, call wire.DatabaseExecServerCall, bh wire.BatchHandle, q string, params []*vom.RawBytes) error { |
| impl := func() error { |
| db := &queryDb{ |
| ctx: ctx, |
| call: call, |
| d: d, |
| bh: bh, |
| sntx: nil, // Filled in later with existing or created sn/tx. |
| ts: nil, // Only filled in if a new batch was created. |
| } |
| st, err := engine.Create(db).PrepareStatement(q) |
| if err != nil { |
| return execCommitOrAbort(db, err) |
| } |
| headers, rs, err := st.Exec(params...) |
| if err != nil { |
| return execCommitOrAbort(db, err) |
| } |
| if rs.Err() != nil { |
| return execCommitOrAbort(db, err) |
| } |
| sender := call.SendStream() |
| // Push the headers first -- the client will retrieve them and return |
| // them separately from the results. |
| var resultHeaders []*vom.RawBytes |
| for _, header := range headers { |
| resultHeaders = append(resultHeaders, vom.RawBytesOf(header)) |
| } |
| sender.Send(resultHeaders) |
| for rs.Advance() { |
| result := rs.Result() |
| if err := sender.Send(result); err != nil { |
| rs.Cancel() |
| return execCommitOrAbort(db, err) |
| } |
| } |
| return execCommitOrAbort(db, rs.Err()) |
| } |
| return impl() |
| } |
| |
| func execCommitOrAbort(qdb *queryDb, err error) error { |
| if qdb.bh != "" { |
| return err // part of an enclosing sn/tx |
| } |
| if err != nil { |
| if qdb.sntx != nil { |
| qdb.sntx.Abort() |
| } |
| return err |
| } else { // err is nil |
| if qdb.ts != nil { |
| return qdb.ts.tx.Commit() |
| } else if qdb.sntx != nil { |
| return qdb.sntx.Abort() |
| } |
| return nil |
| } |
| } |
| |
| func (d *database) SetPermissions(ctx *context.T, call rpc.ServerCall, perms access.Permissions, version string) error { |
| // Permissions checked in d.setPermsInternal, existence in d.s.setDatabasePerms. |
| return d.s.setDatabasePerms(ctx, call, d.id, perms, version) |
| } |
| |
| func (d *database) GetPermissions(ctx *context.T, call rpc.ServerCall) (perms access.Permissions, version string, err error) { |
| allowGetPermissions := []access.Tag{access.Admin} |
| |
| var data DatabaseData |
| impl := func(sntx store.SnapshotOrTransaction) error { |
| // Check perms. |
| _, err := common.GetDataWithAuth(ctx, call, d, allowGetPermissions, d.st, &data) |
| return err |
| } |
| if err := d.runWithNewSnapshot(ctx, call, impl); err != nil { |
| return nil, "", err |
| } |
| return data.Perms, util.FormatVersion(data.Version), nil |
| } |
| |
| func (d *database) GlobChildren__(ctx *context.T, call rpc.GlobChildrenServerCall, matcher *glob.Element) error { |
| allowGlob := []access.Tag{access.Read} |
| |
| impl := func(sntx store.SnapshotOrTransaction) error { |
| // Check perms. |
| if _, err := common.GetPermsWithAuth(ctx, call, d, allowGlob, sntx); err != nil { |
| return err |
| } |
| return util.GlobChildren(ctx, call, matcher, sntx, common.CollectionPermsPrefix) |
| } |
| return d.runWithNewSnapshot(ctx, call, impl) |
| } |
| |
| // See comment in v.io/v23/services/syncbase/service.vdl for why we can't |
| // implement ListCollections using Glob. |
| func (d *database) ListCollections(ctx *context.T, call rpc.ServerCall, bh wire.BatchHandle) ([]wire.Id, error) { |
| allowListCollections := []access.Tag{access.Read} |
| |
| var res []wire.Id |
| impl := func(sntx store.SnapshotOrTransaction) error { |
| // Check perms. |
| if _, err := common.GetPermsWithAuth(ctx, call, d, allowListCollections, sntx); err != nil { |
| return err |
| } |
| it := sntx.Scan(common.ScanPrefixArgs(common.CollectionPermsPrefix, "")) |
| keyBytes := []byte{} |
| for it.Advance() { |
| keyBytes = it.Key(keyBytes) |
| id, err := common.ParseCollectionPermsKey(string(keyBytes)) |
| if err != nil { |
| it.Cancel() |
| return verror.New(verror.ErrInternal, ctx, err) |
| } |
| res = append(res, id) |
| } |
| if err := it.Err(); err != nil { |
| return err |
| } |
| return nil |
| } |
| if err := d.runWithExistingBatchOrNewSnapshot(ctx, call, bh, impl); err != nil { |
| return nil, err |
| } |
| return res, nil |
| } |
| |
| func (d *database) PauseSync(ctx *context.T, call rpc.ServerCall) error { |
| allowPauseSync := []access.Tag{access.Admin} |
| |
| return d.runInNewTransaction(ctx, call, func(ts *transactionState) error { |
| // Check perms. |
| if _, err := common.GetPermsWithAuth(ctx, call, d, allowPauseSync, ts.tx); err != nil { |
| return err |
| } |
| return sbwatchable.AddDbStateChangeRequestOp(ctx, ts.tx, sbwatchable.StateChangePauseSync) |
| }) |
| } |
| |
| func (d *database) ResumeSync(ctx *context.T, call rpc.ServerCall) error { |
| allowResumeSync := []access.Tag{access.Admin} |
| |
| return d.runInNewTransaction(ctx, call, func(ts *transactionState) error { |
| // Check perms. |
| if _, err := common.GetPermsWithAuth(ctx, call, d, allowResumeSync, ts.tx); err != nil { |
| return err |
| } |
| return sbwatchable.AddDbStateChangeRequestOp(ctx, ts.tx, sbwatchable.StateChangeResumeSync) |
| }) |
| } |
| |
| //////////////////////////////////////// |
| // interfaces.Database methods |
| |
| func (d *database) St() *watchable.Store { |
| if !d.exists { |
| vlog.Fatalf("database %v does not exist", d.id) |
| } |
| return d.st |
| } |
| |
| func (d *database) CheckExists(ctx *context.T, call rpc.ServerCall) error { |
| if !d.exists { |
| return d.fuzzyNoExistError(ctx, call) |
| } |
| return nil |
| } |
| |
| func (d *database) Service() interfaces.Service { |
| return d.s |
| } |
| |
| func (d *database) GetCollectionPerms(ctx *context.T, cxId wire.Id, st store.StoreReader) (access.Permissions, error) { |
| if !d.exists { |
| vlog.Fatalf("database %v does not exist", d.id) |
| } |
| c := &collectionReq{ |
| id: cxId, |
| d: d, |
| } |
| var cp interfaces.CollectionPerms |
| err := store.Get(ctx, st, c.permsKey(), &cp) |
| return cp.GetPerms(), err |
| } |
| |
| func (d *database) Id() wire.Id { |
| return d.id |
| } |
| |
| func (d *database) CrConnectionStream() wire.ConflictManagerStartConflictResolverServerStream { |
| d.crMu.Lock() |
| defer d.crMu.Unlock() |
| return d.crStream |
| } |
| |
| func (d *database) ResetCrConnectionStream() { |
| d.crMu.Lock() |
| defer d.crMu.Unlock() |
| // TODO(jlodhia): figure out a way for the connection to gracefully shutdown |
| // so that the client can get an appropriate error msg. |
| d.crStream = nil |
| } |
| |
| //////////////////////////////////////// |
| // query interface implementations |
| |
| // queryDb implements ds.Database. |
| type queryDb struct { |
| ctx *context.T |
| call rpc.ServerCall |
| d *database |
| bh wire.BatchHandle |
| sntx store.SnapshotOrTransaction |
| ts *transactionState // Will only be set if in a transaction (else nil) |
| } |
| |
| func (qdb *queryDb) GetContext() *context.T { |
| return qdb.ctx |
| } |
| |
| func (qdb *queryDb) GetTable(name string, writeAccessReq bool) (ds.Table, error) { |
| // At this point, when the query package calls GetTable with the |
| // writeAccessReq arg, we know whether or not we need a [writable] transaction |
| // or a snapshot. If batchId is already set, there's nothing to do; but if |
| // not, the writeAccessReq arg dictates whether a snapshot or a transaction is |
| // should be created. |
| // TODO(ivanpi): Allow passing in non-default user blessings. |
| userBlessings, _ := security.RemoteBlessingNames(qdb.ctx, qdb.call.Security()) |
| _, user, err := pubutil.AppAndUserPatternFromBlessings(userBlessings...) |
| if err != nil { |
| return nil, err |
| } |
| qt := &queryTable{ |
| qdb: qdb, |
| cReq: &collectionReq{ |
| id: wire.Id{Blessing: string(user), Name: name}, |
| d: qdb.d, |
| }, |
| } |
| if qt.qdb.bh != "" { |
| var err error |
| if writeAccessReq { |
| // We are in a batch (could be snapshot or transaction) |
| // and Write access is required. Attempt to get a |
| // transaction from the request. |
| qt.qdb.ts, err = qt.qdb.d.batchTransaction(qt.qdb.GetContext(), qt.qdb.call, qt.qdb.bh) |
| if err != nil { |
| if verror.ErrorID(err) == wire.ErrReadOnlyBatch.ID { |
| // We are in a snapshot batch, write access cannot be provided. |
| // Return NotWritable. |
| return nil, syncql.NewErrNotWritable(qt.qdb.GetContext(), pubutil.EncodeId(qt.cReq.id)) |
| } |
| return nil, err |
| } |
| qt.qdb.sntx = qt.qdb.ts.tx |
| } else { |
| qt.qdb.sntx, err = qt.qdb.d.batchReader(qt.qdb.GetContext(), qt.qdb.call, qt.qdb.bh) |
| if err != nil { |
| return nil, err |
| } |
| } |
| } else { |
| // Now that we know if write access is required, create a snapshot |
| // or transaction. |
| if !qt.qdb.d.exists { |
| return nil, qt.qdb.d.fuzzyNoExistError(qt.qdb.ctx, qt.qdb.call) |
| } |
| if !writeAccessReq { |
| qt.qdb.sntx = qt.qdb.d.st.NewSnapshot() |
| } else { // writeAccessReq |
| qt.qdb.ts = &transactionState{tx: qt.qdb.d.st.NewWatchableTransaction()} |
| qt.qdb.sntx = qt.qdb.ts.tx |
| } |
| } |
| // Now that we have a collection, we need to check permissions. |
| // Always check for Read access. |
| collectionPerms, err := common.GetPermsWithAuth(qdb.ctx, qdb.call, qt.cReq, []access.Tag{access.Read}, qdb.sntx) |
| if err != nil { |
| return nil, err |
| } |
| if writeAccessReq { |
| // Also check for Write access if requested. |
| if _, err := common.GetPermsWithAuth(qdb.ctx, qdb.call, qt.cReq, []access.Tag{access.Write}, qdb.sntx); err != nil { |
| return nil, err |
| } |
| qt.qdb.ts.MarkDataChanged(qt.cReq.id, collectionPerms) |
| } |
| return qt, nil |
| } |
| |
| // queryTable implements ds.Table. |
| type queryTable struct { |
| qdb *queryDb |
| cReq *collectionReq |
| } |
| |
| func (t *queryTable) GetIndexFields() []ds.Index { |
| // TODO(jkline): If and when secondary indexes are supported, they |
| // would be supplied here. |
| return []ds.Index{} |
| } |
| |
| func (t *queryTable) Delete(k string) (bool, error) { |
| // Create a rowReq and call delete. Permissions will be checked. |
| rowReq := &rowReq{ |
| key: k, |
| c: t.cReq, |
| } |
| if err := rowReq.delete(t.qdb.GetContext(), t.qdb.call, t.qdb.ts); err != nil { |
| return false, err |
| } |
| return true, nil |
| } |
| |
| func (t *queryTable) Scan(indexRanges ...ds.IndexRanges) (ds.KeyValueStream, error) { |
| streams := []store.Stream{} |
| // Syncbase does not currently support secondary indexes. As such, indexRanges |
| // is guaranteed to be one in size as it will only specify the key ranges; |
| // hence, indexRanges[0] below. |
| for _, keyRange := range *indexRanges[0].StringRanges { |
| // TODO(jkline): For now, acquire all of the streams at once to minimize the |
| // race condition. Need a way to Scan multiple ranges at the same state of |
| // uncommitted changes. |
| streams = append(streams, t.qdb.sntx.Scan(common.ScanRangeArgs(common.JoinKeyParts(common.RowPrefix, t.cReq.stKeyPart()), keyRange.Start, keyRange.Limit))) |
| } |
| return &kvs{ |
| t: t, |
| curr: 0, |
| validRow: false, |
| it: streams, |
| err: nil, |
| }, nil |
| } |
| |
| // kvs implements ds.KeyValueStream. |
| type kvs struct { |
| t *queryTable |
| curr int |
| validRow bool |
| currKey string |
| currValue *vom.RawBytes |
| it []store.Stream // array of store.Streams |
| err error |
| } |
| |
| func (s *kvs) Advance() bool { |
| if s.err != nil { |
| return false |
| } |
| for s.curr < len(s.it) { |
| if s.it[s.curr].Advance() { |
| // key |
| keyBytes := s.it[s.curr].Key(nil) |
| parts := common.SplitNKeyParts(string(keyBytes), 3) |
| // TODO(rogulenko): Check access for the key. |
| s.currKey = parts[2] |
| // value |
| valueBytes := s.it[s.curr].Value(nil) |
| var currValue *vom.RawBytes |
| if err := vom.Decode(valueBytes, &currValue); err != nil { |
| s.validRow = false |
| s.err = err |
| s.Cancel() // to cancel iterators after s.curr |
| return false |
| } |
| s.currValue = currValue |
| s.validRow = true |
| return true |
| } |
| // Advance returned false. It could be an err, or it could |
| // be we've reached the end. |
| if err := s.it[s.curr].Err(); err != nil { |
| s.validRow = false |
| s.err = err |
| s.Cancel() // to cancel iterators after s.curr |
| return false |
| } |
| // We've reached the end of the iterator for this keyRange. |
| // Jump to the next one. |
| s.it[s.curr] = nil |
| s.curr++ |
| s.validRow = false |
| } |
| // There are no more prefixes to scan. |
| return false |
| } |
| |
| func (s *kvs) KeyValue() (string, *vom.RawBytes) { |
| if !s.validRow { |
| return "", nil |
| } |
| return s.currKey, s.currValue |
| } |
| |
| func (s *kvs) Err() error { |
| return s.err |
| } |
| |
| func (s *kvs) Cancel() { |
| if s.it != nil { |
| for i := s.curr; i < len(s.it); i++ { |
| s.it[i].Cancel() |
| } |
| s.it = nil |
| } |
| // set curr to end of keyRanges so Advance will return false |
| s.curr = len(s.it) |
| } |
| |
| //////////////////////////////////////// |
| // Authorization hooks |
| |
| var _ common.Permser = (*service)(nil) |
| |
| func (d *database) GetDataWithExistAuth(ctx *context.T, call rpc.ServerCall, st store.StoreReader, v common.PermserData) (parentPerms, perms access.Permissions, _ error) { |
| dd := v.(*DatabaseData) |
| parentPerms, err := common.GetPermsWithExistAndParentResolveAuth(ctx, call, d.s, d.s.st) |
| if err != nil { |
| return nil, nil, err |
| } |
| err = common.GetDataWithExistAuthStep(ctx, call, d.id.String(), parentPerms, st, d.stKey(), dd) |
| return parentPerms, dd.GetPerms(), err |
| } |
| |
| func (d *database) PermserData() common.PermserData { |
| return &DatabaseData{} |
| } |
| |
| //////////////////////////////////////// |
| // Internal helpers |
| |
| func (d *database) stKey() string { |
| return common.DatabasePrefix |
| } |
| |
| func (d *database) fuzzyNoExistError(ctx *context.T, call rpc.ServerCall) error { |
| _, _, err := d.GetDataWithExistAuth(ctx, call, nil, &DatabaseData{}) |
| return err |
| } |
| |
| // Note, the following methods (using batchLookupInternal) handle database |
| // existence checks without leaking existence information. |
| |
| func (d *database) runWithNewSnapshot(ctx *context.T, call rpc.ServerCall, fn func(sntx store.SnapshotOrTransaction) error) error { |
| return d.runWithExistingBatchOrNewSnapshot(ctx, call, "", fn) |
| } |
| |
| func (d *database) runWithExistingBatchOrNewSnapshot(ctx *context.T, call rpc.ServerCall, bh wire.BatchHandle, fn func(sntx store.SnapshotOrTransaction) error) error { |
| if bh != "" { |
| if sntx, err := d.batchReader(ctx, call, bh); err != nil { |
| // Batch does not exist. |
| return err |
| } else { |
| return fn(sntx) |
| } |
| } else { |
| if !d.exists { |
| return d.fuzzyNoExistError(ctx, call) |
| } |
| // Note, prevention of errors leaking existence information relies on the |
| // fact that RunWithSnapshot cannot fail before it calls fn (and therefore |
| // checks access) at least once. |
| return store.RunWithSnapshot(d.st, fn) |
| } |
| } |
| |
| func (d *database) runInNewTransaction(ctx *context.T, call rpc.ServerCall, fn func(ts *transactionState) error) error { |
| return d.runInExistingBatchOrNewTransaction(ctx, call, "", fn) |
| } |
| |
| func (d *database) runInExistingBatchOrNewTransaction(ctx *context.T, call rpc.ServerCall, bh wire.BatchHandle, fn func(ts *transactionState) error) error { |
| if bh != "" { |
| if batch, err := d.batchTransaction(ctx, call, bh); err != nil { |
| // Batch does not exist or is readonly (snapshot). |
| return err |
| } else { |
| return fn(batch) |
| } |
| } else { |
| if !d.exists { |
| return d.fuzzyNoExistError(ctx, call) |
| } |
| // Note, prevention of errors leaking existence information relies on the |
| // fact that runInTransaction cannot fail before it calls fn (and therefore |
| // checks access) at least once. |
| return d.runInTransaction(fn) |
| } |
| } |
| |
| func (d *database) batchReader(ctx *context.T, call rpc.ServerCall, bh wire.BatchHandle) (store.SnapshotOrTransaction, error) { |
| sn, ts, _, err := d.batchLookupInternal(ctx, call, bh) |
| if err != nil { |
| return nil, err |
| } |
| if sn != nil { |
| return sn, nil |
| } |
| return ts.tx, nil |
| } |
| |
| func (d *database) batchTransaction(ctx *context.T, call rpc.ServerCall, bh wire.BatchHandle) (*transactionState, error) { |
| sn, ts, _, err := d.batchLookupInternal(ctx, call, bh) |
| if err != nil { |
| return nil, err |
| } |
| if sn != nil { |
| return nil, wire.NewErrReadOnlyBatch(ctx) |
| } |
| return ts, nil |
| } |
| |
| // batchLookupInternal parses the batch handle and retrieves the corresponding |
| // snapshot or transaction. It returns an error if the handle is malformed or |
| // the batch does not exist. Otherwise, exactly one of sn and ts will be != nil. |
| // If a non-nil error is returned, it will not leak database existence if the |
| // caller is not authorized to know it. |
| func (d *database) batchLookupInternal(ctx *context.T, call rpc.ServerCall, bh wire.BatchHandle) (sn store.Snapshot, ts *transactionState, batchId uint64, _ error) { |
| if bh == "" { |
| return nil, nil, 0, verror.New(verror.ErrInternal, ctx, "batch lookup for empty handle") |
| } |
| bType, bId, err := common.SplitBatchHandle(bh) |
| if err != nil { |
| return nil, nil, 0, err |
| } |
| if !d.exists { |
| return nil, nil, 0, d.fuzzyNoExistError(ctx, call) |
| } |
| d.mu.Lock() |
| defer d.mu.Unlock() |
| var found bool |
| switch bType { |
| case common.BatchTypeSn: |
| sn, found = d.sns[bId] |
| case common.BatchTypeTx: |
| ts, found = d.txs[bId] |
| } |
| if !found { |
| _, _, err := d.GetDataWithExistAuth(ctx, call, d.st, &DatabaseData{}) |
| if err == nil { |
| return nil, nil, bId, wire.NewErrUnknownBatch(ctx) |
| } |
| return nil, nil, bId, err |
| } |
| return sn, ts, bId, nil |
| } |
| |
| func (d *database) setPermsInternal(ctx *context.T, call rpc.ServerCall, perms access.Permissions, version string) error { |
| allowSetPermissions := []access.Tag{access.Admin} |
| |
| if !d.exists { |
| vlog.Fatalf("database %v does not exist", d.id) |
| } |
| if err := common.ValidatePerms(ctx, perms, wire.AllDatabaseTags); err != nil { |
| return err |
| } |
| return store.RunInTransaction(d.st, func(tx store.Transaction) error { |
| var data DatabaseData |
| if _, err := common.GetDataWithAuth(ctx, call, d, allowSetPermissions, tx, &data); err != nil { |
| return err |
| } |
| if err := util.CheckVersion(ctx, version, data.Version); err != nil { |
| return err |
| } |
| data.Perms = perms |
| data.Version++ |
| return store.Put(ctx, tx, d.stKey(), &data) |
| }) |
| } |
| |
| // runInTransaction runs the given fn in a transaction, managing retries and |
| // commit/abort. |
| func (d *database) runInTransaction(fn func(ts *transactionState) error) error { |
| // TODO(rogulenko): Change the default number of attempts to 3. Currently, |
| // some storage engine tests fail when the number of attempts is that low. |
| return d.runInTransactionWithOpts(&store.TransactionOptions{NumAttempts: 100}, fn) |
| } |
| |
| // runInTransactionWithOpts runs the given fn in a transaction, managing retries |
| // and commit/abort. |
| func (d *database) runInTransactionWithOpts(opts *store.TransactionOptions, fn func(ts *transactionState) error) error { |
| var err error |
| for i := 0; i < opts.NumAttempts; i++ { |
| // TODO(sadovsky): Should NewTransaction return an error? If not, how will |
| // we deal with RPC errors when talking to remote storage engines? (Note, |
| // client-side BeginBatch returns an error.) |
| ts := &transactionState{tx: d.st.NewWatchableTransaction()} |
| if err = fn(ts); err != nil { |
| ts.tx.Abort() |
| return err |
| } |
| // TODO(sadovsky): Commit() can fail for a number of reasons, e.g. RPC |
| // failure or ErrConcurrentTransaction. Depending on the cause of failure, |
| // it may be desirable to retry the Commit() and/or to call Abort(). |
| if err = ts.tx.Commit(); verror.ErrorID(err) != store.ErrConcurrentTransaction.ID { |
| return err |
| } |
| } |
| return err |
| } |