X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/b9f0177e6a477a518fd5a89156fafde57f2dddf8..74e03b2190fea52b3ea5800c50eca55b6e50b220:/services/ws/session_v0.go diff --git a/services/ws/session_v0.go b/services/ws/session_v0.go index 2b108a2b5c..4fbfc489cf 100644 --- a/services/ws/session_v0.go +++ b/services/ws/session_v0.go @@ -1,3 +1,7 @@ +// Copyright (C) The Arvados Authors. All rights reserved. +// +// SPDX-License-Identifier: AGPL-3.0 + package main import ( @@ -16,7 +20,16 @@ var ( errQueueFull = errors.New("client queue full") errFrameTooBig = errors.New("frame too big") - sendObjectAttributes = []string{"state", "name"} + // Send clients only these keys from the + // log.properties.old_attributes and + // log.properties.new_attributes hashes. + sendObjectAttributes = []string{ + "is_trashed", + "name", + "owner_uuid", + "portable_data_hash", + "state", + } v0subscribeOK = []byte(`{"status":200}`) v0subscribeFail = []byte(`{"status":400}`) @@ -35,6 +48,9 @@ type v0session struct { setupOnce sync.Once } +// newSessionV0 returns a v0 session: a partial port of the Rails/puma +// implementation, with just enough functionality to support Workbench +// and arv-mount. func newSessionV0(ws wsConn, sendq chan<- interface{}, db *sql.DB, pc permChecker, ac *arvados.Client) (session, error) { sess := &v0session{ sendq: sendq, @@ -83,7 +99,17 @@ func (sess *v0session) EventMessage(e *event) ([]byte, error) { return nil, nil } - ok, err := sess.permChecker.Check(detail.ObjectUUID) + var permTarget string + if detail.EventType == "delete" { + // It's pointless to check permission by reading + // ObjectUUID if it has just been deleted, but if the + // client has permission on the parent project then + // it's OK to send the event. + permTarget = detail.ObjectOwnerUUID + } else { + permTarget = detail.ObjectUUID + } + ok, err := sess.permChecker.Check(permTarget) if err != nil || !ok { return nil, err } @@ -97,6 +123,7 @@ func (sess *v0session) EventMessage(e *event) ([]byte, error) { "object_owner_uuid": detail.ObjectOwnerUUID, "object_kind": kind, "event_type": detail.EventType, + "event_at": detail.EventAt, } if detail.Properties != nil && detail.Properties["text"] != nil { msg["properties"] = detail.Properties @@ -135,7 +162,7 @@ func (sub *v0subscribe) sendOldEvents(sess *v0session) { if sub.LastLogID == 0 { return } - sess.log.WithField("LastLogID", sub.LastLogID).Debug("getOldEvents") + sess.log.WithField("LastLogID", sub.LastLogID).Debug("sendOldEvents") // Here we do a "select id" query and queue an event for every // log since the given ID, then use (*event)Detail() to // retrieve the whole row and decide whether to send it. This @@ -150,16 +177,26 @@ func (sub *v0subscribe) sendOldEvents(sess *v0session) { sub.LastLogID, time.Now().UTC().Add(-10*time.Minute).Format(time.RFC3339Nano)) if err != nil { - sess.log.WithError(err).Error("db.Query failed") + sess.log.WithError(err).Error("sendOldEvents db.Query failed") return } + + var ids []uint64 for rows.Next() { var id uint64 err := rows.Scan(&id) if err != nil { - sess.log.WithError(err).Error("row Scan failed") + sess.log.WithError(err).Error("sendOldEvents row Scan failed") continue } + ids = append(ids, id) + } + if err := rows.Err(); err != nil { + sess.log.WithError(err).Error("sendOldEvents db.Query failed") + } + rows.Close() + + for _, id := range ids { for len(sess.sendq)*2 > cap(sess.sendq) { // Ugly... but if we fill up the whole client // queue with a backlog of old events, a @@ -168,6 +205,10 @@ func (sub *v0subscribe) sendOldEvents(sess *v0session) { // client will probably reconnect and do the // same thing all over again. time.Sleep(100 * time.Millisecond) + if sess.ws.Request().Context().Err() != nil { + // Session terminated while we were sleeping + return + } } now := time.Now() e := &event{ @@ -184,9 +225,6 @@ func (sub *v0subscribe) sendOldEvents(sess *v0session) { } } } - if err := rows.Err(); err != nil { - sess.log.WithError(err).Error("db.Query failed") - } } type v0subscribe struct {